wip

pulp_container/app/models.py

@@ -145,7 +145,7 @@ def init_labels(self):
         if self.config_blob.data:
             # if config-blob is stored in the database
             blob_data = json.loads(self.config_blob.data)
-            config = blob_data.get("config",{})
+            config = blob_data.get("config", {})
             if "Labels" in config.keys():
                 self.labels = config["Labels"] or {}
         else:

pulp_container/app/registry_api.py

@@ -1008,9 +1008,8 @@ def handle_safe_method(self, request, path, pk):
                 raise BlobNotFound(digest=pk)
 
         if blob.data:
-            media_type = BLOB_CONTENT_TYPE
             headers = {
-                "Content-Type": media_type,
+                "Content-Type": BLOB_CONTENT_TYPE,
                 "Docker-Content-Digest": pk,
                 "Docker-Distribution-API-Version": "registry/2.0",
             }

pulp_container/app/tasks/builder.py

@@ -12,7 +12,7 @@
     Manifest,
     Tag,
 )
-from pulp_container.constants import MEDIA_TYPE
+from pulp_container.constants import DB_BLOB_SIZE, MEDIA_TYPE
 from pulpcore.plugin.models import Artifact, ContentArtifact, Content
 
 
@@ -35,10 +35,10 @@ def get_or_create_blob(layer_json, manifest, path):
     except Blob.DoesNotExist:
         layer_file_name = os.path.join(path, layer_json["digest"][7:])
         blob = Blob(digest=layer_json["digest"])
-        if layer_json["mediaType"] in [MEDIA_TYPE.CONFIG_BLOB_OCI, MEDIA_TYPE.CONFIG_BLOB]:
-            with open(layer_file_name, "r") as content_file:
-                raw_data = content_file.read()
-            blob.data = raw_data.encode("utf-8")
+
+        if os.path.getsize(layer_file_name) < DB_BLOB_SIZE:
+            with open(layer_file_name, "rb") as content_file:
+                blob.data = content_file.read()
             blob.save()
         else:
             layer_artifact = Artifact.init_and_validate(layer_file_name)

pulp_container/app/tasks/sync_stages.py

@@ -13,6 +13,7 @@
 from pulpcore.plugin.stages import DeclarativeArtifact, DeclarativeContent, Stage, ContentSaver
 
 from pulp_container.constants import (
+    DB_BLOB_SIZE,
     MEDIA_TYPE,
     SIGNATURE_API_EXTENSION_VERSION,
     SIGNATURE_HEADER,
@@ -350,12 +351,12 @@ async def handle_blobs(self, manifest_dc, content_data):
         for layer in content_data.get("layers") or content_data.get("fsLayers"):
             if not self._include_layer(layer):
                 continue
-            blob_dc = self.create_blob(layer)
+            blob_dc = await self.create_blob(layer)
             manifest_dc.extra_data["blob_dcs"].append(blob_dc)
             await self.put(blob_dc)
         config = content_data.get("config", None)
         if config:
-            blob_dc = await self.create_config_blob(config)
+            blob_dc = await self.create_blob(config, False, False)
             manifest_dc.extra_data["config_blob_dc"] = blob_dc
             await self.put(blob_dc)
 
@@ -518,49 +519,70 @@ async def create_listed_manifest(self, manifest_data):
         )
         return {"manifest_dc": man_dc, "platform": platform, "content_data": content_data}
 
-    def create_blob(self, blob_data, deferred_download=True):
+    async def create_blob(self, blob_data, deferred_download=True, is_layer=True):
         """
         Create blob.
 
         Args:
             blob_data (dict): Data about a blob
             deferred_download (bool): boolean that indicates whether not to download a blob
                 immediatly. Config blob is downloaded regardless of the remote's settings
+            is_layer (bool): Boolean to identify if this is a layer blob (the size should be
+                retrived from the manifest) or a config blob
 
         """
         digest = blob_data.get("digest") or blob_data.get("blobSum")
-        blob_artifact = Artifact(sha256=digest[len("sha256:") :])
-        blob = Blob(digest=digest)
         relative_url = "/v2/{name}/blobs/{digest}".format(
             name=self.remote.namespaced_upstream_name, digest=digest
         )
         blob_url = urljoin(self.remote.url, relative_url)
-        da = DeclarativeArtifact(
-            artifact=blob_artifact,
-            url=blob_url,
-            relative_path=digest,
-            remote=self.remote,
-            deferred_download=deferred_download and self.deferred_download,
-        )
-        blob_dc = DeclarativeContent(content=blob, d_artifacts=[da])
+        blob_size, raw_data = await self._get_blob_size_data(blob_data, blob_url, is_layer)
+
+        if blob_size > DB_BLOB_SIZE:
+            blob_artifact = Artifact(sha256=digest[len("sha256:") :])
+            blob = Blob(digest=digest)
+            da = DeclarativeArtifact(
+                artifact=blob_artifact,
+                url=blob_url,
+                relative_path=digest,
+                remote=self.remote,
+                deferred_download=deferred_download and self.deferred_download,
+            )
+            blob_dc = DeclarativeContent(content=blob, d_artifacts=[da])
+        else:
+            # if this is a layer with size < DB_BLOB_SIZE, we will need to download it
+            if is_layer:
+                downloader = self.remote.get_downloader(url=blob_url)
+                response = await downloader.run(extra_data={"headers": V2_ACCEPT_HEADERS})
+                with open(response.path, "rb") as content_file:
+                    raw_data = content_file.read()
+            artifactless_blob = Blob(digest=digest, data=raw_data)
+            blob_dc = DeclarativeContent(content=artifactless_blob)
 
         return blob_dc
 
-    async def create_config_blob(self, blob_data):
-        digest = blob_data.get("digest")
-        relative_url = "/v2/{name}/blobs/{digest}".format(
-            name=self.remote.namespaced_upstream_name, digest=digest
-        )
-        blob_url = urljoin(self.remote.url, relative_url)
+    async def _get_blob_size_data(self, blob_data, blob_url, is_layer):
+        """
+        Return the Blob Size and its content.
 
-        downloader = self.remote.get_downloader(url=blob_url)
-        response = await downloader.run(extra_data={"headers": V2_ACCEPT_HEADERS})
-        with open(response.path, "rb") as content_file:
-            raw_data = content_file.read()
-        config_blob = Blob(digest=digest, data=raw_data)
-        config_blob_dc = DeclarativeContent(content=config_blob)
+        If this is a layer blob, the size is a mandatory field in the manifest.
+        If this is a config blob, we will need to first download it to discover its size and, since
+        the file content will now be stored in the database, it is returned to avoid a new download.
 
-        return config_blob_dc
+        Args:
+            blob_data (dict): Data about a blob
+            blob_url (str): The remote url of blob
+            is_layer (bool): Boolean to identify if this is a layer blob (the size should be
+                retrived from manifest) or a config blob
+        """
+        if is_layer:
+            return blob_data.get("size"), None
+        else:
+            downloader = self.remote.get_downloader(url=blob_url)
+            response = await downloader.run(extra_data={"headers": V2_ACCEPT_HEADERS})
+            with open(response.path, "rb") as content_file:
+                raw_data = content_file.read()
+            return len(raw_data), raw_data
 
     async def create_signatures(self, man_dc, signature_source):
         """