Gzip stream

.gitignore

@@ -10,5 +10,6 @@ ENV/
 # pycharm proj
 .idea
 
-# stray tar.gz files used for manual testing
+# stray archive files used for manual testing
+*.tar
 *.tar.gz

core/__init__.py

@@ -1,4 +1,4 @@
-from .manifest_creator import ImageManifestCreator
+from .image_manifest_creator import ImageManifestCreator
 from .registry import Registry
 from .extractor import Extractor
 from .processor import Processor

core/extractor.py

@@ -6,7 +6,7 @@
 import humanfriendly
 
 
-class Extractor(object):
+class Extractor:
     def __init__(self, logger, archive_path):
         self._logger = logger.get_child('tar')
         self._archive_path = os.path.abspath(archive_path)

core/image_manifest_creator.py

@@ -0,0 +1,28 @@
+import json
+
+
+class ImageManifestCreator:
+    def __init__(self, config_path, layers_info, config_info):
+        self._config_path = config_path
+        self._layers_info = layers_info
+        self._config_info = config_info
+
+    def create(self):
+        manifest = dict()
+        manifest["schemaVersion"] = 2
+        manifest["mediaType"] = "application/vnd.docker.distribution.manifest.v2+json"
+        manifest["config"] = {
+            "mediaType": "application/vnd.docker.container.image.v1+json",
+            "size": self._config_info['size'],
+            "digest": self._config_info['digest'],
+        }
+        manifest["layers"] = []
+        for layer_info in self._layers_info:
+            layer_data = {
+                "mediaType": "application/vnd.docker.image.rootfs.diff.tar.gzip",
+                "size": layer_info['size'],
+                "digest": layer_info['digest'],
+            }
+            manifest["layers"].append(layer_data)
+
+        return json.dumps(manifest)

core/processor.py

@@ -2,6 +2,7 @@
 import multiprocessing.pool
 import time
 import os.path
+import shutil
 import json
 
 import humanfriendly
@@ -10,14 +11,17 @@
 from . import extractor
 
 
-class Processor(object):
+class Processor:
     def __init__(
         self,
         logger,
+        tmp_dir,
+        tmp_dir_override,
         parallel,
         registry_url,
         archive_path,
         stream=False,
+        gzip_layers=False,
         login=None,
         password=None,
         ssl_verify=True,
@@ -26,6 +30,8 @@ def __init__(
     ):
         self._logger = logger
         self._parallel = parallel
+        self._tmp_dir = tmp_dir
+        self._tmp_dir_override = tmp_dir_override
 
         if parallel > 1 and stream:
             self._logger.info(
@@ -36,6 +42,7 @@ def __init__(
 
         self._registry = registry.Registry(
             logger=self._logger,
+            gzip_layers=gzip_layers,
             registry_url=registry_url,
             stream=stream,
             login=login,
@@ -55,8 +62,14 @@ def process(self):
         """
         start_time = time.time()
         results = []
-        with tempfile.TemporaryDirectory() as tmp_dir_name:
-
+        if self._tmp_dir_override:
+            tmp_dir_name = self._tmp_dir_override
+            os.mkdir(tmp_dir_name, 0o700)
+        else:
+            tmp_dir_name = tempfile.mkdtemp(dir=self._tmp_dir)
+
+        # since we're not always using TemporaryDirectory we're making and cleaning up ourselves
+        try:
             self._logger.info(
                 'Processing archive',
                 archive_path=self._extractor.archive_path,
@@ -83,9 +96,12 @@ def process(self):
                 pool.close()
                 pool.join()
 
-        # this will throw if any pool worker caught an exception
-        for res in results:
-            res.get()
+            # this will throw if any pool worker caught an exception
+            for res in results:
+                res.get()
+        finally:
+            shutil.rmtree(tmp_dir_name)
+            self._logger.verbose('Removed workdir', tmp_dir_name=tmp_dir_name)
 
         elapsed = time.time() - start_time
         self._logger.info(
@@ -95,11 +111,16 @@ def process(self):
         )
 
     @staticmethod
-    def _get_manifest(tmp_dir_name):
-        with open(os.path.join(tmp_dir_name, 'manifest.json'), 'r') as fh:
+    def _get_manifest(archive_dir):
+        with open(os.path.join(archive_dir, 'manifest.json'), 'r') as fh:
             manifest = json.loads(fh.read())
             return manifest
 
+    @staticmethod
+    def _write_manifest(archive_dir, contents):
+        with open(os.path.join(archive_dir, 'manifest.json'), 'w') as fh:
+            json.dump(contents, fh)
+
 
 #
 # Global wrappers to use with multiprocessing.pool.Pool which can't pickle instance methods