You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@buildstream.apache.org by no...@apache.org on 2020/12/29 12:41:55 UTC
[buildstream] branch jennis/alternative_remote_expiry created (now
bb2a10b)
This is an automated email from the ASF dual-hosted git repository.
not-in-ldap pushed a change to branch jennis/alternative_remote_expiry
in repository https://gitbox.apache.org/repos/asf/buildstream.git.
at bb2a10b pushreceive.py: Abstract the buffer_
This branch includes the following new commits:
new fcea521 _ostree.py: Reintroduce remove()
new 5f6841a _ostree.py: Introduce _list_all_refs() and list_artifacts()
new 62343da pushreceive.py: Ensure there is a repopath
new fc0c8d3 pushreceive.py: Remove LRP artifacts from cache, introduce clean_up_cache()
new 07f9e21 element_generators.py: Add a utility function to create arbitrary size elements
new da17a86 push.py: Add tests to test remote cache expiry
new 2b9ad60 pushreceive.py: Add comments to code and name variables more sensibly
new 79da729 pushreceive.py: Ensure huge artifacts are not pushed
new bb2a10b pushreceive.py: Abstract the buffer_
The 9 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] 06/09: push.py: Add tests to test remote cache expiry
Posted by no...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
not-in-ldap pushed a commit to branch jennis/alternative_remote_expiry
in repository https://gitbox.apache.org/repos/asf/buildstream.git
commit da17a86f9a47b3c13da84e6d9e290b88a589d407
Author: James Ennis <ja...@codethink.com>
AuthorDate: Wed Apr 25 10:47:21 2018 +0100
push.py: Add tests to test remote cache expiry
The tests include a new environment variable: BST_CACHE_QUOTA.
This variable is used in receive_main() in pushreceive.py.
Test names: test_artifact_expires, test_large_artifact_is_not_cached_remotely
test_recently_pulled_artifact_does_not_expire: marked as xfail until
we implement LRU expiry in remote share
---
tests/frontend/push.py | 185 ++++++++++++++++++++++++++++++++++++++++++++++++-
1 file changed, 184 insertions(+), 1 deletion(-)
diff --git a/tests/frontend/push.py b/tests/frontend/push.py
index e9c2daa..3111a04 100644
--- a/tests/frontend/push.py
+++ b/tests/frontend/push.py
@@ -1,8 +1,11 @@
import os
import shutil
import pytest
+from collections import namedtuple
+from unittest.mock import MagicMock
+
from buildstream._exceptions import ErrorDomain
-from tests.testutils import cli, create_artifact_share
+from tests.testutils import cli, create_artifact_share, create_element_size
from tests.testutils.site import IS_LINUX
# Project directory
@@ -195,3 +198,183 @@ def test_push_after_pull(cli, tmpdir, datafiles):
result.assert_success()
assert result.get_pulled_elements() == ['target.bst']
assert result.get_pushed_elements() == ['target.bst']
+
+
+# Ensure that when an artifact's size exceeds available disk space
+# the least recently pushed artifact is deleted in order to make room for
+# the incoming artifact.
+@pytest.mark.datafiles(DATA_DIR)
+def test_artifact_expires(cli, datafiles, tmpdir):
+ project = os.path.join(datafiles.dirname, datafiles.basename)
+ element_path = os.path.join(project, 'elements')
+
+ # Create an artifact share (remote artifact cache) in the tmpdir/artifactshare
+ share = create_artifact_share(os.path.join(str(tmpdir), 'artifactshare'))
+
+ # Mock the os.statvfs() call to return a named tuple which emulates an
+ # os.statvfs_result object
+ statvfs_result = namedtuple('statvfs_result', 'f_blocks f_bfree f_bsize')
+ os.statvfs = MagicMock(return_value=statvfs_result(f_blocks=int(10e9),
+ f_bfree=(int(12e6) + int(2e9)),
+ f_bsize=1))
+
+ # Configure bst to push to the cache
+ cli.configure({
+ 'artifacts': {'url': share.repo, 'push': True},
+ })
+
+ # Create and build an element of 5 MB
+ create_element_size('element1.bst', element_path, [], int(5e6)) # [] => no deps
+ result = cli.run(project=project, args=['build', 'element1.bst'])
+ result.assert_success()
+
+ # Create and build an element of 5 MB
+ create_element_size('element2.bst', element_path, [], int(5e6)) # [] => no deps
+ result = cli.run(project=project, args=['build', 'element2.bst'])
+ result.assert_success()
+
+ # update the share
+ share.update_summary()
+
+ # check that element's 1 and 2 are cached both locally and remotely
+ assert cli.get_element_state(project, 'element1.bst') == 'cached'
+ assert_shared(cli, share, project, 'element1.bst')
+ assert cli.get_element_state(project, 'element2.bst') == 'cached'
+ assert_shared(cli, share, project, 'element2.bst')
+
+ # update mocked available disk space now that two 5 MB artifacts have been added
+ os.statvfs = MagicMock(return_value=statvfs_result(f_blocks=int(10e9),
+ f_bfree=(int(2e6) + int(2e9)),
+ f_bsize=1))
+
+ # Create and build another element of 5 MB (This will exceed the free disk space available)
+ create_element_size('element3.bst', element_path, [], int(5e6))
+ result = cli.run(project=project, args=['build', 'element3.bst'])
+ result.assert_success()
+
+ # update the share
+ share.update_summary()
+
+ # Ensure it is cached both locally and remotely
+ assert cli.get_element_state(project, 'element3.bst') == 'cached'
+ assert_shared(cli, share, project, 'element3.bst')
+
+ # Ensure element1 has been removed from the share
+ assert_not_shared(cli, share, project, 'element1.bst')
+ # Ensure that elemen2 remains
+ assert_shared(cli, share, project, 'element2.bst')
+
+
+# Test that a large artifact, whose size exceeds the quota, is not pushed
+# to the remote share
+@pytest.mark.xfail
+@pytest.mark.datafiles(DATA_DIR)
+def test_artifact_too_large(cli, datafiles, tmpdir):
+ project = os.path.join(datafiles.dirname, datafiles.basename)
+ element_path = os.path.join(project, 'elements')
+
+ # Create an artifact share (remote cache) in tmpdir/artifactshare
+ share = create_artifact_share(os.path.join(str(tmpdir), 'artifactshare'))
+
+ # Mock a file system with 5 MB total space
+ statvfs_result = namedtuple('statvfs_result', 'f_blocks f_bfree f_bsize')
+ os.statvfs = MagicMock(return_value=statvfs_result(f_blocks=int(5e6) + int(2e9),
+ f_bfree=(int(5e6) + int(2e9)),
+ f_bsize=1))
+
+ # Configure bst to push to the remote cache
+ cli.configure({
+ 'artifacts': {'url': share.repo, 'push': True},
+ })
+
+ # Create and push a 3MB element
+ create_element_size('small_element.bst', element_path, [], int(3e6))
+ result = cli.run(project=project, args=['build', 'small_element.bst'])
+ result.assert_success()
+
+ # Create and try to push a 6MB element.
+ create_element_size('large_element.bst', element_path, [], int(6e6))
+ result = cli.run(project=project, args=['build', 'large_element.bst'])
+ result.assert_success()
+
+ # update the cache
+ share.update_summary()
+
+ # Ensure that the small artifact is still in the share
+ assert cli.get_element_state(project, 'small_element.bst') == 'cached'
+ assert_shared(cli, share, project, 'small_element.bst')
+
+ # Ensure that the artifact is cached locally but NOT remotely
+ assert cli.get_element_state(project, 'large_element.bst') == 'cached'
+ assert_not_shared(cli, share, project, 'large_element.bst')
+
+
+# Test that when an element is pulled recently, it is not considered the LRU element.
+# NOTE: We expect this test to fail as the current implementation of remote cache
+# expiry only expiries from least recently pushed. NOT least recently used. This will
+# hopefully change when we implement as CAS cache.
+@pytest.mark.xfail
+@pytest.mark.datafiles(DATA_DIR)
+def test_recently_pulled_artifact_does_not_expire(cli, datafiles, tmpdir):
+ project = os.path.join(datafiles.dirname, datafiles.basename)
+ element_path = os.path.join(project, 'elements')
+
+ # Create an artifact share (remote cache) in tmpdir/artifactshare
+ share = create_artifact_share(os.path.join(str(tmpdir), 'artifactshare'))
+
+ # Mock a file system with 12 MB free disk space
+ statvfs_result = namedtuple('statvfs_result', 'f_blocks f_bfree f_bsize')
+ os.statvfs = MagicMock(return_value=statvfs_result(f_blocks=int(10e9) + int(2e9),
+ f_bfree=(int(12e6) + int(2e9)),
+ f_bsize=1))
+
+ # Configure bst to push to the cache
+ cli.configure({
+ 'artifacts': {'url': share.repo, 'push': True},
+ })
+
+ # Create and build 2 elements, each of 5 MB.
+ create_element_size('element1.bst', element_path, [], int(5e6))
+ result = cli.run(project=project, args=['build', 'element1.bst'])
+ result.assert_success()
+
+ create_element_size('element2.bst', element_path, [], int(5e6))
+ result = cli.run(project=project, args=['build', 'element2.bst'])
+ result.assert_success()
+
+ share.update_summary()
+
+ # Ensure they are cached locally
+ assert cli.get_element_state(project, 'element1.bst') == 'cached'
+ assert cli.get_element_state(project, 'element2.bst') == 'cached'
+
+ # Ensure that they have been pushed to the cache
+ assert_shared(cli, share, project, 'element1.bst')
+ assert_shared(cli, share, project, 'element2.bst')
+
+ # Remove element1 from the local cache
+ cli.remove_artifact_from_cache(project, 'element1.bst')
+ assert cli.get_element_state(project, 'element1.bst') != 'cached'
+
+ # Pull the element1 from the remote cache (this should update its mtime)
+ result = cli.run(project=project, args=['pull', 'element1.bst', '--remote',
+ share.repo])
+ result.assert_success()
+
+ # Ensure element1 is cached locally
+ assert cli.get_element_state(project, 'element1.bst') == 'cached'
+
+ # Create and build the element3 (of 5 MB)
+ create_element_size('element3.bst', element_path, [], int(5e6))
+ result = cli.run(project=project, args=['build', 'element3.bst'])
+ result.assert_success()
+
+ share.update_summary()
+
+ # Make sure it's cached locally and remotely
+ assert cli.get_element_state(project, 'element3.bst') == 'cached'
+ assert_shared(cli, share, project, 'element3.bst')
+
+ # Ensure that element2 was deleted from the share and element1 remains
+ assert_not_shared(cli, share, project, 'element2.bst')
+ assert_shared(cli, share, project, 'element1.bst')
[buildstream] 02/09: _ostree.py: Introduce _list_all_refs() and
list_artifacts()
Posted by no...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
not-in-ldap pushed a commit to branch jennis/alternative_remote_expiry
in repository https://gitbox.apache.org/repos/asf/buildstream.git
commit 5f6841a101dd386c0f02eb869f78e3bfbc0054f8
Author: Tristan Maat <tm...@tlater.net>
AuthorDate: Sat Mar 24 17:56:56 2018 +0000
_ostree.py: Introduce _list_all_refs() and list_artifacts()
The unused list_remote_refs() function has also been removed as
part of this commit.
---
buildstream/_ostree.py | 44 ++++++++++++++++++++++++++++++++++++++++++++
1 file changed, 44 insertions(+)
diff --git a/buildstream/_ostree.py b/buildstream/_ostree.py
index 246b54f..e40df5f 100644
--- a/buildstream/_ostree.py
+++ b/buildstream/_ostree.py
@@ -552,3 +552,47 @@ def configure_remote(repo, remote, url, key_url=None):
repo.remote_gpg_import(remote, stream, None, 0, None)
except GLib.GError as e:
raise OSTreeError("Failed to add gpg key from url '{}': {}".format(key_url, e.message)) from e
+
+
+# list_artifacts():
+#
+# List cached artifacts in Least Recently Modified (LRM) order.
+#
+# Returns:
+# (list) - A list of refs in LRM order
+#
+def list_artifacts(repo):
+ # string of: /path/to/repo/refs/heads
+ ref_heads = os.path.join(repo.get_path().get_path(), 'refs', 'heads')
+
+ # obtain list of <project>/<element>/<key>
+ refs = _list_all_refs(repo).keys()
+
+ mtimes = []
+ for ref in refs:
+ ref_path = os.path.join(ref_heads, ref)
+ if os.path.exists(ref_path):
+ # Obtain the mtime (the time a file was last modified)
+ mtimes.append(os.path.getmtime(ref_path))
+
+ # NOTE: Sorted will sort from earliest to latest, thus the
+ # first element of this list will be the file modified earliest.
+ return [ref for _, ref in sorted(zip(mtimes, refs))]
+
+
+# _list_all_refs():
+#
+# Create a list of all refs.
+#
+# Args:
+# repo (OSTree.Repo): The repo
+#
+# Returns:
+# (dict): A dict of refs to checksums.
+#
+def _list_all_refs(repo):
+ try:
+ _, refs = repo.list_refs(None)
+ return refs
+ except GLib.GError as e:
+ raise OSTreeError(message=e.message) from e
[buildstream] 01/09: _ostree.py: Reintroduce remove()
Posted by no...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
not-in-ldap pushed a commit to branch jennis/alternative_remote_expiry
in repository https://gitbox.apache.org/repos/asf/buildstream.git
commit fcea521334e766d1f40a489ddf09ecf8cb9a1014
Author: Tristan Maat <tm...@tlater.net>
AuthorDate: Fri Mar 23 17:26:32 2018 +0000
_ostree.py: Reintroduce remove()
---
buildstream/_ostree.py | 36 ++++++++++++++++++++++++++++++++++++
1 file changed, 36 insertions(+)
diff --git a/buildstream/_ostree.py b/buildstream/_ostree.py
index dfa7567..246b54f 100644
--- a/buildstream/_ostree.py
+++ b/buildstream/_ostree.py
@@ -225,6 +225,42 @@ def exists(repo, ref):
return has_object
+# remove():
+#
+# Removes the given commit or symbolic ref from the repo.
+#
+# Args:
+# repo (OSTree.Repo): The repo
+# ref (str): A commit checksum or symbolic ref
+# defer_prune (bool): Whether to defer pruning to the caller. NOTE:
+# The space won't be freed until you manually
+# call repo.prune.
+#
+# Returns:
+# (int|None) The amount of space pruned from the repository in
+# Bytes, or None if defer_prune is True
+#
+def remove(repo, ref, *, defer_prune=False):
+
+ # Get the commit checksum, this will:
+ #
+ # o Return a commit checksum if ref is a symbolic branch
+ # o Return the same commit checksum if ref is a valid commit checksum
+ # o Return None if the ostree repo doesnt know this ref.
+ #
+ check = checksum(repo, ref)
+ if check is None:
+ raise OSTreeError("Could not find artifact for ref '{}'".format(ref))
+
+ repo.set_ref_immediate(None, ref, None)
+
+ if not defer_prune:
+ _, _, _, pruned = repo.prune(OSTree.RepoPruneFlags.REFS_ONLY, -1)
+ return pruned
+
+ return None
+
+
# checksum():
#
# Returns the commit checksum for a given symbolic ref,
[buildstream] 04/09: pushreceive.py: Remove LRP artifacts from
cache, introduce clean_up_cache()
Posted by no...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
not-in-ldap pushed a commit to branch jennis/alternative_remote_expiry
in repository https://gitbox.apache.org/repos/asf/buildstream.git
commit fc0c8d30a5b4890da7fe087a29ddbb7b13816e08
Author: James Ennis <ja...@codethink.com>
AuthorDate: Wed Apr 18 15:01:37 2018 +0100
pushreceive.py: Remove LRP artifacts from cache, introduce clean_up_cache()
---
buildstream/_artifactcache/pushreceive.py | 57 +++++++++++++++++++++++++++++--
1 file changed, 54 insertions(+), 3 deletions(-)
diff --git a/buildstream/_artifactcache/pushreceive.py b/buildstream/_artifactcache/pushreceive.py
index 209aef6..f8fd76d 100644
--- a/buildstream/_artifactcache/pushreceive.py
+++ b/buildstream/_artifactcache/pushreceive.py
@@ -33,6 +33,7 @@ from urllib.parse import urlparse
import click
import gi
+from .. import _ostree
from .. import _signals # nopep8
from .._profile import Topics, profile_start, profile_end
@@ -277,10 +278,14 @@ class PushMessageReader(object):
_, args = self.receive([PushCommandType.update])
return args
- def receive_putobjects(self, repo):
-
+ def receive_putobjects(self, repo, repopath):
received_objects = []
+ # Determine the available disk space, in bytes, of the file system
+ # which mounts the repo
+ stats = os.statvfs(repopath)
+ free_disk_space = stats.f_bfree * stats.f_bsize
+
# Open a TarFile for reading uncompressed tar from a stream
tar = tarfile.TarFile.open(mode='r|', fileobj=self.file)
@@ -288,6 +293,7 @@ class PushMessageReader(object):
#
# This should block while tar.next() reads the next
# tar object from the stream.
+ buffer_ = int(2e9)
while True:
filepos = tar.fileobj.tell()
tar_info = tar.next()
@@ -300,7 +306,16 @@ class PushMessageReader(object):
tar.fileobj.read(512)
break
+ # obtain size of tar object in bytes
+ artifact_size = tar_info.size
+
+ if artifact_size > free_disk_space - buffer_:
+ # Clean up the cache with a buffer of 2GB
+ removed_size = clean_up_cache(repo, artifact_size, free_disk_space, buffer_)
+ free_disk_space += removed_size
+
tar.extract(tar_info, self.tmpdir)
+ free_disk_space -= artifact_size
received_objects.append(tar_info.name)
# Finished with this stream
@@ -645,7 +660,7 @@ class OSTreeReceiver(object):
return 0
# Receive the actual objects
- received_objects = self.reader.receive_putobjects(self.repo)
+ received_objects = self.reader.receive_putobjects(self.repo, self.repopath)
# Ensure that pusher has sent all objects
self.reader.receive_done()
@@ -794,6 +809,42 @@ def push(repo, remote, branches, output):
return False
+# clean_up_cache()
+#
+# Keep removing Least Recently Pushed (LRP) artifacts in a cache until there
+# is enough space for the incoming artifact
+#
+# Args:
+# repo: OSTree.Repo object
+# free_disk_space: The available disk space on the file system in bytes
+# artifact_size: The size of the artifact in bytes
+# buffer_: The amount of headroom we want on disk.
+#
+# Returns:
+# int: The total bytes removed on the filesystem
+#
+def clean_up_cache(repo, artifact_size, free_disk_space, buffer_):
+ # obtain a list of LRP artifacts
+ LRP_artifacts = _ostree.list_artifacts(repo)
+
+ removed_size = 0 # in bytes
+ while artifact_size - removed_size > free_disk_space - buffer_:
+ try:
+ to_remove = LRP_artifacts.pop(0) # The first element in the list is the LRP artifact
+ except IndexError:
+ logging.info("There are no more artifacts left in the cache. Adding artifact...")
+ break
+
+ removed_size += _ostree.remove(repo, to_remove, defer_prune=False)
+
+ if removed_size > 0:
+ logging.info("Successfully removed {} bytes from the cache".format(removed_size))
+ else:
+ logging.info("No artifacts were removed from the cache.")
+
+ return removed_size
+
+
@click.command(short_help="Receive pushed artifacts over ssh")
@click.option('--verbose', '-v', is_flag=True, default=False, help="Verbose mode")
@click.option('--debug', '-d', is_flag=True, default=False, help="Debug mode")
[buildstream] 09/09: pushreceive.py: Abstract the buffer_
Posted by no...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
not-in-ldap pushed a commit to branch jennis/alternative_remote_expiry
in repository https://gitbox.apache.org/repos/asf/buildstream.git
commit bb2a10b975fd5ad8287dae483467da1015d5c888
Author: James Ennis <ja...@codethink.com>
AuthorDate: Mon Jun 4 11:02:08 2018 +0100
pushreceive.py: Abstract the buffer_
---
buildstream/_artifactcache/pushreceive.py | 17 ++++++++---------
1 file changed, 8 insertions(+), 9 deletions(-)
diff --git a/buildstream/_artifactcache/pushreceive.py b/buildstream/_artifactcache/pushreceive.py
index ab74e80..d4b378c 100644
--- a/buildstream/_artifactcache/pushreceive.py
+++ b/buildstream/_artifactcache/pushreceive.py
@@ -289,8 +289,9 @@ class PushMessageReader(object):
# Determine the available disk space, in bytes, of the file system
# which mounts the repo
stats = os.statvfs(repopath)
- free_disk_space = stats.f_bfree * stats.f_bsize
- total_disk_space = stats.f_blocks * stats.f_bsize
+ buffer_ = int(2e9) # Add a 2 GB buffer
+ free_disk_space = (stats.f_bfree * stats.f_bsize) - buffer_
+ total_disk_space = (stats.f_blocks * stats.f_bsize) - buffer_
# Open a TarFile for reading uncompressed tar from a stream
tar = tarfile.TarFile.open(mode='r|', fileobj=self.file)
@@ -299,7 +300,6 @@ class PushMessageReader(object):
#
# This should block while tar.next() reads the next
# tar object from the stream.
- buffer_ = int(2e9)
while True:
filepos = tar.fileobj.tell()
tar_info = tar.next()
@@ -315,14 +315,14 @@ class PushMessageReader(object):
# obtain size of tar object in bytes
artifact_size = tar_info.size
- if artifact_size > total_disk_space - buffer_:
+ if artifact_size > total_disk_space:
raise ArtifactTooLargeException("Artifact of size: {} is too large for "
"the filesystem which mounts the remote "
"cache".format(artifact_size))
- if artifact_size > free_disk_space - buffer_:
+ if artifact_size > free_disk_space:
# Clean up the cache with a buffer of 2GB
- removed_size = clean_up_cache(repo, artifact_size, free_disk_space, buffer_)
+ removed_size = clean_up_cache(repo, artifact_size, free_disk_space)
free_disk_space += removed_size
tar.extract(tar_info, self.tmpdir)
@@ -853,17 +853,16 @@ def push(repo, remote, branches, output):
# repo: OSTree.Repo object
# free_disk_space: The available disk space on the file system in bytes
# artifact_size: The size of the artifact in bytes
-# buffer_: The amount of headroom we want on disk.
#
# Returns:
# int: The total bytes removed on the filesystem
#
-def clean_up_cache(repo, artifact_size, free_disk_space, buffer_):
+def clean_up_cache(repo, artifact_size, free_disk_space):
# obtain a list of LRP artifacts
LRP_artifacts = _ostree.list_artifacts(repo)
removed_size = 0 # in bytes
- while artifact_size - removed_size > free_disk_space - buffer_:
+ while artifact_size - removed_size > free_disk_space:
try:
to_remove = LRP_artifacts.pop(0) # The first element in the list is the LRP artifact
except IndexError:
[buildstream] 07/09: pushreceive.py: Add comments to code and name
variables more sensibly
Posted by no...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
not-in-ldap pushed a commit to branch jennis/alternative_remote_expiry
in repository https://gitbox.apache.org/repos/asf/buildstream.git
commit 2b9ad60f9a486186e52dd5c36b7263088e77a8cf
Author: James Ennis <ja...@codethink.com>
AuthorDate: Fri Apr 20 17:20:35 2018 +0100
pushreceive.py: Add comments to code and name variables more sensibly
---
buildstream/_artifactcache/pushreceive.py | 14 +++++++++++++-
1 file changed, 13 insertions(+), 1 deletion(-)
diff --git a/buildstream/_artifactcache/pushreceive.py b/buildstream/_artifactcache/pushreceive.py
index f8fd76d..e17e1de 100644
--- a/buildstream/_artifactcache/pushreceive.py
+++ b/buildstream/_artifactcache/pushreceive.py
@@ -113,7 +113,7 @@ class PushMessageWriter(object):
def __init__(self, file, byteorder=sys.byteorder):
self.file = file
self.byteorder = byteorder
- self.msg_byteorder = python_to_msg_byteorder(self.byteorder)
+ self.msg_byteorder = python_to_msg_byteorder(self.byteorder) # 'l' or 'B'
def encode_header(self, cmdtype, size):
header = self.msg_byteorder.encode() + \
@@ -388,6 +388,7 @@ def foo_run(func, args, stdin_fd, stdout_fd, stderr_fd):
class ProcessWithPipes(object):
def __init__(self, func, args, *, stderr=None):
+ # Create a pipe and return a pair of file descriptors (r, w)
r0, w0 = os.pipe()
r1, w1 = os.pipe()
if stderr is None:
@@ -432,10 +433,12 @@ class OSTreePusher(object):
# Enumerate branches to push
if branches is None:
+ # obtain a dict of 'refs': 'checksums'
_, self.refs = self.repo.list_refs(None, None)
else:
self.refs = {}
for branch in branches:
+ # branch is a ref, now find its checksum (i.e. rev)
_, rev = self.repo.resolve_rev(branch, False)
self.refs[branch] = rev
@@ -454,6 +457,9 @@ class OSTreePusher(object):
logging.info('Executing {}'.format(' '.join(ssh_cmd)))
if self.remote_host:
+ # subprocess.Popen(args, bufsize=-1,...)
+ # Executes a child program in a new process which returns an open file
+ # object connected to the pipe.
self.ssh = subprocess.Popen(ssh_cmd, stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=self.output,
@@ -530,6 +536,8 @@ class OSTreePusher(object):
for branch, rev in self.refs.items():
remote_rev = remote_refs.get(branch, '0' * 64)
if rev != remote_rev:
+ # if the checksums for a branch aren't equal add a tuple of
+ # the remote_rev and local rev to a new dictionary.
update_refs[branch] = remote_rev, rev
if not update_refs:
logging.info('Nothing to update')
@@ -550,9 +558,12 @@ class OSTreePusher(object):
commits = set()
exc_info = None
ref_count = 0
+
+ # update the remote checksum with the local one
for branch, revs in update_refs.items():
logging.info('Updating {} {} to {}'.format(branch, revs[0], revs[1]))
try:
+ # obtain a set of the commits needed to be pushed
self.needed_commits(revs[0], revs[1], commits)
ref_count += 1
except PushExistsException:
@@ -564,6 +575,7 @@ class OSTreePusher(object):
raise exc_info[0].with_traceback(exc_info[1], exc_info[2])
logging.info('Enumerating objects to send')
+ # obtain a set of the objects which need to be pushed to the server
objects = self.needed_objects(commits)
# Send all the objects to receiver, checking status after each
[buildstream] 03/09: pushreceive.py: Ensure there is a repopath
Posted by no...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
not-in-ldap pushed a commit to branch jennis/alternative_remote_expiry
in repository https://gitbox.apache.org/repos/asf/buildstream.git
commit 62343da07bf2be74ebcec1dcdcc7d03ae9bce3bf
Author: James Ennis <ja...@codethink.com>
AuthorDate: Wed Apr 18 12:42:52 2018 +0100
pushreceive.py: Ensure there is a repopath
---
buildstream/_artifactcache/pushreceive.py | 3 +++
1 file changed, 3 insertions(+)
diff --git a/buildstream/_artifactcache/pushreceive.py b/buildstream/_artifactcache/pushreceive.py
index ecedc66..209aef6 100644
--- a/buildstream/_artifactcache/pushreceive.py
+++ b/buildstream/_artifactcache/pushreceive.py
@@ -576,6 +576,9 @@ class OSTreeReceiver(object):
if self.repopath is None:
self.repo = OSTree.Repo.new_default()
+ self.repopath = self.repo.get_path().get_path()
+ # NOTE: OSTree.Repo.get_path() returns Gio.File
+ # Gio.File.get_path() returns a string of the pathway
else:
self.repo = OSTree.Repo.new(Gio.File.new_for_path(self.repopath))
self.repo.open(None)
[buildstream] 05/09: element_generators.py: Add a utility function
to create arbitrary size elements
Posted by no...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
not-in-ldap pushed a commit to branch jennis/alternative_remote_expiry
in repository https://gitbox.apache.org/repos/asf/buildstream.git
commit 07f9e2116143b6b44af9ea660d828734d08e01e9
Author: James Ennis <ja...@codethink.com>
AuthorDate: Tue Apr 24 16:38:02 2018 +0100
element_generators.py: Add a utility function to create arbitrary size elements
New file consisting of functions that allow us to generate elements
on the fly
__init__.py now includes create_element_size()
---
tests/testutils/__init__.py | 1 +
tests/testutils/element_generators.py | 40 +++++++++++++++++++++++++++++++++++
2 files changed, 41 insertions(+)
diff --git a/tests/testutils/__init__.py b/tests/testutils/__init__.py
index 7e5b792..93143b5 100644
--- a/tests/testutils/__init__.py
+++ b/tests/testutils/__init__.py
@@ -1,3 +1,4 @@
from .runcli import cli, cli_integration
from .repo import create_repo, ALL_REPO_KINDS
from .artifactshare import create_artifact_share
+from .element_generators import create_element_size
diff --git a/tests/testutils/element_generators.py b/tests/testutils/element_generators.py
new file mode 100644
index 0000000..3f6090d
--- /dev/null
+++ b/tests/testutils/element_generators.py
@@ -0,0 +1,40 @@
+import os
+
+from buildstream import _yaml
+
+
+# create_element_size()
+#
+# This will open a "<name>_data" file for writing and write
+# <size> MB of urandom (/dev/urandom) "stuff" into the file.
+# A bst import element file is then created: <name>.bst
+#
+# Args:
+# name: (str) of the element name (e.g. target.bst)
+# path: (str) pathway to the project/elements directory
+# dependencies: A list of strings (can also be an empty list)
+# size: (int) size of the element in bytes
+#
+# Returns:
+# Nothing (creates a .bst file of specified size)
+#
+def create_element_size(name, path, dependencies, size):
+ os.makedirs(path, exist_ok=True)
+
+ # Create a file to be included in this element's artifact
+ with open(os.path.join(path, name + '_data'), 'wb+') as f:
+ f.write(os.urandom(size))
+
+ # Simplest case: We want this file (of specified size) to just
+ # be an import element.
+ element = {
+ 'kind': 'import',
+ 'sources': [
+ {
+ 'kind': 'local',
+ 'path': os.path.join(path, name + '_data')
+ }
+ ],
+ 'depends': dependencies
+ }
+ _yaml.dump(element, os.path.join(path, name))
[buildstream] 08/09: pushreceive.py: Ensure huge artifacts are not
pushed
Posted by no...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
not-in-ldap pushed a commit to branch jennis/alternative_remote_expiry
in repository https://gitbox.apache.org/repos/asf/buildstream.git
commit 79da7295f71ce8ac6f31b9ed47617dbd0401b63d
Author: James Ennis <ja...@codethink.com>
AuthorDate: Wed May 30 15:56:41 2018 +0100
pushreceive.py: Ensure huge artifacts are not pushed
---
buildstream/_artifactcache/pushreceive.py | 37 ++++++++++++++++++++++++++-----
tests/frontend/push.py | 1 -
2 files changed, 32 insertions(+), 6 deletions(-)
diff --git a/buildstream/_artifactcache/pushreceive.py b/buildstream/_artifactcache/pushreceive.py
index e17e1de..ab74e80 100644
--- a/buildstream/_artifactcache/pushreceive.py
+++ b/buildstream/_artifactcache/pushreceive.py
@@ -56,6 +56,11 @@ class PushExistsException(Exception):
pass
+# Trying to push an artifact that is too large
+class ArtifactTooLargeException(Exception):
+ pass
+
+
class PushCommandType(Enum):
info = 0
update = 1
@@ -285,6 +290,7 @@ class PushMessageReader(object):
# which mounts the repo
stats = os.statvfs(repopath)
free_disk_space = stats.f_bfree * stats.f_bsize
+ total_disk_space = stats.f_blocks * stats.f_bsize
# Open a TarFile for reading uncompressed tar from a stream
tar = tarfile.TarFile.open(mode='r|', fileobj=self.file)
@@ -309,6 +315,11 @@ class PushMessageReader(object):
# obtain size of tar object in bytes
artifact_size = tar_info.size
+ if artifact_size > total_disk_space - buffer_:
+ raise ArtifactTooLargeException("Artifact of size: {} is too large for "
+ "the filesystem which mounts the remote "
+ "cache".format(artifact_size))
+
if artifact_size > free_disk_space - buffer_:
# Clean up the cache with a buffer of 2GB
removed_size = clean_up_cache(repo, artifact_size, free_disk_space, buffer_)
@@ -579,7 +590,13 @@ class OSTreePusher(object):
objects = self.needed_objects(commits)
# Send all the objects to receiver, checking status after each
- self.writer.send_putobjects(self.repo, objects)
+ try:
+ self.writer.send_putobjects(self.repo, objects)
+ except BrokenPipeError:
+ # If the remote closes, we receive a BrokenPipeError
+ # Return 1 to notify the frontend that something went
+ # wrong on the server.
+ return 1
# Inform receiver that all objects have been sent
self.writer.send_done()
@@ -626,8 +643,11 @@ class OSTreeReceiver(object):
def run(self):
try:
exit_code = self.do_run()
- self.close()
- return exit_code
+ except ArtifactTooLargeException:
+ logging.warning("The artifact was too large for the filesystem which mounts "
+ "the remote cache.")
+ exit_code = 0
+
except:
# BLIND EXCEPT - Just abort if we receive any exception, this
# can be a broken pipe, a tarfile read error when the remote
@@ -635,6 +655,9 @@ class OSTreeReceiver(object):
self.close()
raise
+ self.close()
+ return exit_code
+
def do_run(self):
# Receive remote info
args = self.reader.receive_info()
@@ -844,8 +867,12 @@ def clean_up_cache(repo, artifact_size, free_disk_space, buffer_):
try:
to_remove = LRP_artifacts.pop(0) # The first element in the list is the LRP artifact
except IndexError:
- logging.info("There are no more artifacts left in the cache. Adding artifact...")
- break
+ # This exception is caught if there are no more artifacts in the list
+ # LRP_artifacts. This means the the artifact is too large for the filesystem
+ # so we abort the process
+ raise ArtifactTooLargeException("Artifact of size {} is too large for "
+ "the filesystem which mounts the remote "
+ "cache".format(artifact_size))
removed_size += _ostree.remove(repo, to_remove, defer_prune=False)
diff --git a/tests/frontend/push.py b/tests/frontend/push.py
index 3111a04..3d40c7d 100644
--- a/tests/frontend/push.py
+++ b/tests/frontend/push.py
@@ -267,7 +267,6 @@ def test_artifact_expires(cli, datafiles, tmpdir):
# Test that a large artifact, whose size exceeds the quota, is not pushed
# to the remote share
-@pytest.mark.xfail
@pytest.mark.datafiles(DATA_DIR)
def test_artifact_too_large(cli, datafiles, tmpdir):
project = os.path.join(datafiles.dirname, datafiles.basename)