cascache: move tmp directory one level up

The CAS uses a temp directory while manipulating the cache, temporary
files can be added and removed while adding artifacts to the cache here.

Since calculation of the cache size happens in parallel to artifact
cache additions, this causes race conditions in the size calculation
job, as we end up calling `stat` on temporary files which are being
removed in parallel.

Handle this by moving the temporary directory out of the way, and avoid
considering the tmp directory when calculating the cache size

buildstream/_artifactcache/artifactcache.py

@@ -80,6 +80,8 @@ class ArtifactCache():
         self.context = context
         self.required_artifacts = set()
         self.extractdir = os.path.join(context.artifactdir, 'extract')
+        self.tmpdir = os.path.join(context.artifactdir, 'tmp')
+
         self.max_size = context.cache_quota
         self.estimated_size = None
 
@@ -89,7 +91,8 @@ class ArtifactCache():
         self._local = False
         self.cache_size = None
 
-        os.makedirs(context.artifactdir, exist_ok=True)
+        os.makedirs(self.extractdir, exist_ok=True)
+        os.makedirs(self.tmpdir, exist_ok=True)
 
     ################################################
     #  Methods implemented on the abstract class   #

buildstream/_artifactcache/cascache.py

@@ -56,7 +56,8 @@ class CASCache(ArtifactCache):
         super().__init__(context)
 
         self.casdir = os.path.join(context.artifactdir, 'cas')
-        os.makedirs(os.path.join(self.casdir, 'tmp'), exist_ok=True)
+        os.makedirs(os.path.join(self.casdir, 'refs', 'heads'), exist_ok=True)
+        os.makedirs(os.path.join(self.casdir, 'objects'), exist_ok=True)
 
         self._enable_push = enable_push
 
@@ -85,8 +86,6 @@ class CASCache(ArtifactCache):
             # artifact has already been extracted
             return dest
 
-        os.makedirs(self.extractdir, exist_ok=True)
-
         with tempfile.TemporaryDirectory(prefix='tmp', dir=self.extractdir) as tmpdir:
             checkoutdir = os.path.join(tmpdir, ref)
             self._checkout(checkoutdir, tree)
@@ -394,7 +393,7 @@ class CASCache(ArtifactCache):
         try:
             h = hashlib.sha256()
             # Always write out new file to avoid corruption if input file is modified
-            with tempfile.NamedTemporaryFile(dir=os.path.join(self.casdir, 'tmp')) as out:
+            with tempfile.NamedTemporaryFile(dir=self.tmpdir) as out:
                 # Set mode bits to 0644
                 os.chmod(out.name, stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP | stat.S_IROTH)
 
@@ -764,7 +763,7 @@ class CASCache(ArtifactCache):
             # already in local cache
             return
 
-        with tempfile.NamedTemporaryFile(dir=os.path.join(self.casdir, 'tmp')) as out:
+        with tempfile.NamedTemporaryFile(dir=self.tmpdir) as out:
             self._fetch_blob(remote, tree, out)
 
             directory = remote_execution_pb2.Directory()
@@ -778,7 +777,7 @@ class CASCache(ArtifactCache):
                     # already in local cache
                     continue
 
-                with tempfile.NamedTemporaryFile(dir=os.path.join(self.casdir, 'tmp')) as f:
+                with tempfile.NamedTemporaryFile(dir=self.tmpdir) as f:
                     self._fetch_blob(remote, filenode.digest, f)
 
                     digest = self.add_object(path=f.name)

buildstream/_artifactcache/casserver.py

@@ -161,7 +161,7 @@ class _ByteStreamServicer(bytestream_pb2_grpc.ByteStreamServicer):
         offset = 0
         finished = False
         resource_name = None
-        with tempfile.NamedTemporaryFile(dir=os.path.join(self.cas.casdir, 'tmp')) as out:
+        with tempfile.NamedTemporaryFile(dir=self.cas.tmpdir) as out:
             for request in request_iterator:
                 assert not finished
                 assert request.write_offset == offset