Merge pull request #2875 from yueguobin/feat/docker-image-sync

Sync Docker images from the controller to remote computes

gns3server/api/routes/compute/docker_nodes.py

@@ -81,6 +81,7 @@ async def create_docker_node(project_id: UUID, node_data: schemas.DockerCreate)
         extra_configs=node_data.get("extra_configs"),
         memory=node_data.get("memory", 0),
         cpus=node_data.get("cpus", 0),
+        image_digest=node_data.get("image_digest"),
     )
     # Pop keys already consumed by create_node above so the setattr
     # fallback loop below only applies truly extra keys and does not
@@ -89,7 +90,7 @@ async def create_docker_node(project_id: UUID, node_data: schemas.DockerCreate)
         "console", "console_type", "console_resolution", "console_http_port",
         "console_http_path", "aux", "aux_type", "start_command", "environment",
         "adapters", "mac_address", "extra_hosts", "extra_volumes", "extra_configs",
-        "memory", "cpus",
+        "memory", "cpus", "image_digest",
     ):
         node_data.pop(key, None)
     for name, value in node_data.items():

gns3server/api/routes/compute/images.py

@@ -53,6 +53,16 @@ async def pull_docker_image(image: str = Body(..., embed=True, min_length=1, pat
     await docker_manager.pull_image(image, force=True)
 
 
+@router.post("/docker/images/load", status_code=status.HTTP_204_NO_CONTENT)
+async def load_docker_image(request: Request) -> None:
+    """
+    Load a Docker image into the Docker daemon from a docker save tar stream.
+    """
+
+    docker_manager = Docker.instance()
+    await docker_manager.load_image(request.stream())
+
+
 @router.get("/dynamips/images")
 async def get_dynamips_images() -> List[dict]:
     """

gns3server/compute/docker/__init__.py

@@ -283,7 +283,12 @@ class Docker(BaseManager):
         :returns: HTTP response
         """
 
-        data = json.dumps(data)
+        if isinstance(data, dict):
+            data = json.dumps(data)
+            headers = {"content-type": "application/json"}
+        else:
+            # not a dict (e.g. a Docker image tar stream): let aiohttp stream the raw body
+            headers = {"content-type": "application/x-tar"}
         if timeout is None:
             timeout = 60 * 60 * 24 * 31  # One month timeout
 
@@ -301,9 +306,7 @@ class Docker(BaseManager):
                 url,
                 params=params,
                 data=data,
-                headers={
-                    "content-type": "application/json",
-                },
+                headers=headers,
                 timeout=timeout,
             )
         except aiohttp.ClientError as e:
@@ -410,6 +413,58 @@ class Docker(BaseManager):
         if progress_callback:
             progress_callback(f"Success pulling image {image}")
 
+    @locking
+    async def load_image(self, stream, progress_callback=None):
+        """
+        Load a Docker image into the Docker daemon from a docker save tar stream
+
+        :param stream: An async iterable of bytes (the tar produced by docker save)
+        :param progress_callback: A function that receive a log message about image load progress
+        """
+
+        if progress_callback:
+            progress_callback("Loading Docker image from stream")
+        response = await self.http_query("POST", "images/load", data=stream, timeout=None)
+        # The load api will stream status via an HTTP JSON stream
+        content = ""
+        try:
+            while True:
+                try:
+                    chunk = await response.content.read(CHUNK_SIZE)
+                except aiohttp.ServerDisconnectedError as e:
+                    raise DockerError("Disconnected while loading Docker image") from e
+                except asyncio.TimeoutError as e:
+                    raise DockerError("Timeout while loading Docker image") from e
+                if not chunk:
+                    break
+                content += chunk.decode("utf-8", errors="ignore")
+
+                try:
+                    while True:
+                        content = content.lstrip(" \r\n\t")
+                        answer, index = json.JSONDecoder().raw_decode(content)
+                        if not isinstance(answer, dict):
+                            raise DockerError("Invalid response while loading Docker image")
+                        error_detail = answer.get("errorDetail")
+                        error = answer.get("error")
+                        if not error and isinstance(error_detail, dict):
+                            error = error_detail.get("message")
+                        if error:
+                            raise DockerError(error)
+                        if "stream" in answer and progress_callback:
+                            progress_callback(answer["stream"].rstrip())
+                        content = content[index:]
+                except ValueError:  # Partial JSON
+                    pass
+
+            if content.strip():
+                raise DockerError("Invalid response while loading Docker image")
+        finally:
+            response.close()
+
+        if progress_callback:
+            progress_callback("Docker image loaded")
+
     async def list_images(self):
         """
         Gets Docker image list.

gns3server/compute/docker/docker_vm.py

@@ -47,6 +47,7 @@ from ..base_node import BaseNode
 from ..adapters.ethernet_adapter import EthernetAdapter
 from ..nios.nio_udp import NIOUDP
 from .docker_error import DockerError, DockerHttp304Error, DockerHttp404Error, DockerHttp409Error
+from ..error import ImageMissingError
 
 import logging
 
@@ -121,6 +122,7 @@ class DockerVM(BaseNode):
         extra_configs=None,
         memory=0,
         cpus=0,
+        image_digest=None,
     ):
 
         if not is_rfc1123_hostname_valid(name):
@@ -134,6 +136,10 @@ class DockerVM(BaseNode):
         if ":" not in image:
             image = f"{image}:latest"
         self._image = image
+        # image id expected by the controller (set when the image is available on
+        # the controller host): a different local image under the same tag is
+        # treated as missing so the controller re-syncs the expected version
+        self._image_digest = image_digest
         # assign through the property setters so creation and updates apply
         # the same value normalization (e.g. "" -> None)
         self.start_command = start_command
@@ -574,13 +580,24 @@ class DockerVM(BaseNode):
         try:
             image_infos = await self._get_image_information()
         except DockerHttp404Error:
-            log.info("Image '{}' is missing, pulling it from Docker repository...".format(self._image))
-            await self.pull_image(self._image)
-            image_infos = await self._get_image_information()
+            # the image is not on the local Docker daemon: raise ImageMissingError so the
+            # controller can sync it (docker save -> load from the controller host, or pull)
+            # and retry the node creation
+            raise ImageMissingError(self._image)
 
         if image_infos is None:
             raise DockerError(f"Cannot get information for image '{self._image}', please try again.")
 
+        if self._image_digest and image_infos.get("Id") != self._image_digest:
+            # the tag exists but points to a different image (e.g. a moved :latest):
+            # report it as missing so the controller re-syncs the expected version
+            local_id = image_infos.get("Id")
+            log.info(
+                f"Image '{self._image}' version mismatch on this compute: "
+                f"local id '{local_id}' != expected '{self._image_digest}'"
+            )
+            raise ImageMissingError(self._image)
+
         available_cpus = psutil.cpu_count(logical=True)
         if self._cpus > available_cpus:
             raise DockerError(
@@ -1943,16 +1960,6 @@ class DockerVM(BaseNode):
             )
         )
 
-    async def pull_image(self, image):
-        """
-        Pulls an image from Docker repository
-        """
-
-        def callback(msg):
-            self.project.emit("log.info", {"message": msg})
-
-        await self.manager.pull_image(image, progress_callback=callback)
-
     async def _start_ubridge_capture(self, adapter_number, output_file, port_number=0):
         """
         Starts a packet capture in uBridge.

gns3server/controller/node.py

@@ -465,6 +465,7 @@ class Node:
         data["node_id"] = self._id
         if self._node_type == "docker":
             timeout = None
+            await self._add_docker_image_digest(data)
         else:
             timeout = 1200
         trial = 0
@@ -770,6 +771,9 @@ class Node:
         if the image exists
         """
 
+        if self._node_type == "docker":
+            return await self._sync_missing_docker_image(img)
+
         for directory in images_directories(type):
             image = os.path.join(directory, img)
             if os.path.exists(image):
@@ -785,6 +789,66 @@ class Node:
                 return True
         return False
 
+    async def _add_docker_image_digest(self, data):
+        """
+        Pin the Docker image of a node to the image id found on the Docker daemon
+        of the controller host, so that a compute holding a different image under
+        the same tag (e.g. a moved :latest) gets it re-synced instead of silently
+        reusing the stale copy. Not set when the image is not on the controller host.
+        """
+
+        from gns3server.compute.docker import Docker
+        from gns3server.compute.docker.docker_error import DockerError
+
+        image = data.get("image")
+        if not image:
+            return
+        try:
+            image_info = await Docker.instance().query("GET", f"images/{image}/json")
+        except DockerError:
+            # not available on the controller host: nothing to pin against
+            return
+        image_id = image_info.get("Id")
+        if image_id:
+            data["image_digest"] = image_id
+
+    async def _sync_missing_docker_image(self, image):
+        """
+        Export a Docker image from the Docker daemon on the controller host and
+        stream it to the remote compute, or ask the compute to pull it when the
+        image is not available locally
+        """
+
+        from gns3server.compute.docker import Docker
+        from gns3server.compute.docker.docker_error import DockerError
+
+        try:
+            response = await Docker.instance().http_query("GET", f"images/{image}/get", timeout=None)
+        except DockerError:
+            # the image is not on the Docker daemon of the controller host: ask the
+            # compute to pull it from the Docker repository as a fallback
+            self.project.emit_notification(
+                "log.info",
+                {"message": f"Docker image '{image}' is not on the controller host, "
+                            f"asking compute '{self._compute.name}' to pull it"}
+            )
+            await self._compute.post("/docker/images/pull", data={"image": image}, timeout=None)
+            return True
+
+        self.project.emit_notification(
+            "log.info",
+            {"message": f"Syncing Docker image '{image}' to compute '{self._compute.name}'"}
+        )
+        try:
+            await self._compute.post("/docker/images/load", data=response.content, timeout=None)
+        finally:
+            response.close()
+        self.project.emit_notification(
+            "log.info",
+            {"message": f"Docker image '{image}' has been synced to compute '{self._compute.name}'"}
+        )
+        return True
+
     async def dynamips_auto_idlepc(self):
         """
         Compute the idle PC for a dynamips node

gns3server/schemas/compute/docker_nodes.py

@@ -75,6 +75,12 @@ class DockerCreate(DockerBase):
     application_id: Optional[int] = Field(
         None, ge=1, le=1022, description="IOL application ID for iol-runner images (allocated by the controller)"
     )
+    image_digest: Optional[str] = Field(
+        None, pattern=r"^sha256:[a-f0-9]{64}$",
+        description="Image id the controller expects for 'image' (sha256:<hex>, resolved from the Docker "
+                    "daemon on the controller host). When set and the compute holds a different image "
+                    "under the same tag, the image is reported as missing so the controller re-syncs it"
+    )
 
 
 class DockerUpdate(DockerBase):

tests/api/routes/compute/test_images.py

@@ -47,3 +47,16 @@ class TestImagesRoutes:
             )
             mock.assert_not_called()
             assert response.status_code == status.HTTP_422_UNPROCESSABLE_CONTENT
+
+    async def test_load_docker_image(self, app: FastAPI, compute_client: AsyncClient) -> None:
+
+        with asyncio_patch("gns3server.compute.docker.Docker.load_image") as mock:
+            response = await compute_client.post(
+                app.url_path_for("compute:load_docker_image"),
+                content=b"docker-save-tar-bytes"
+            )
+            mock.assert_called_once()
+            # the tar is streamed to the Docker daemon as an async iterable
+            stream = mock.call_args[0][0]
+            assert hasattr(stream, "__aiter__")
+            assert response.status_code == status.HTTP_204_NO_CONTENT

tests/compute/docker/test_docker.py

@@ -255,6 +255,55 @@ async def test_pull_image_propagates_timeout():
             response.close.assert_called_once()
 
 
+@pytest.mark.asyncio
+async def test_load_image():
+
+    class Content:
+
+        def __init__(self):
+            self._chunks = [b'{"stream": "Loaded image ID: sha256:e90e34656806"}', b""]
+
+        async def read(self, size):
+            return self._chunks.pop(0)
+
+    response = MagicMock()
+    response.content = Content()
+
+    async def tar_stream():
+        yield b"tar-bytes"
+
+    stream = tar_stream()
+    messages = []
+    with asyncio_patch("gns3server.compute.docker.Docker.http_query", return_value=response) as mock:
+        await Docker.instance().load_image(stream, progress_callback=messages.append)
+        mock.assert_called_with("POST", "images/load", data=stream, timeout=None)
+    response.close.assert_called_once()
+    assert any("Loaded image" in message for message in messages)
+
+
+@pytest.mark.asyncio
+async def test_load_image_error():
+
+    class Content:
+
+        def __init__(self):
+            self._chunks = [b'{"error": "invalid tar file"}', b""]
+
+        async def read(self, size):
+            return self._chunks.pop(0)
+
+    response = MagicMock()
+    response.content = Content()
+
+    async def tar_stream():
+        yield b"not-a-tar"
+
+    with asyncio_patch("gns3server.compute.docker.Docker.http_query", return_value=response):
+        with pytest.raises(DockerError, match="invalid tar file"):
+            await Docker.instance().load_image(tar_stream())
+        response.close.assert_called_once()
+
+
 @pytest.mark.asyncio
 async def test_docker_check_connection_docker_minimum_version(vm):
 

tests/compute/docker/test_docker_vm.py

@@ -30,6 +30,7 @@ from gns3server.compute.ubridge.ubridge_error import UbridgeNamespaceError
 from gns3server.compute.compute_error import ComputeError
 from gns3server.compute.docker.docker_vm import DockerVM
 from gns3server.compute.docker.docker_error import DockerError, DockerHttp404Error
+from gns3server.compute.error import ImageMissingError
 from gns3server.compute.docker import Docker
 
 
@@ -555,65 +556,44 @@ async def test_create_environment_with_last_new_line_character(compute_project,
 @pytest.mark.asyncio
 async def test_create_image_not_available(compute_project, manager):
 
-    call = 0
-    async def information():
-        nonlocal call
-        if call == 0:
-            call += 1
-            raise DockerHttp404Error("missing")
-        else:
-            return {}
+    vm = DockerVM("test", str(uuid.uuid4()), compute_project, manager, "ubuntu")
+    vm._get_image_information = MagicMock(side_effect=DockerHttp404Error("missing"))
+    with asyncio_patch("gns3server.compute.docker.Docker.query") as query_mock:
+        with pytest.raises(ImageMissingError, match="ubuntu:latest"):
+            await vm.create()
+        query_mock.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_create_image_digest_match(compute_project, manager):
 
     response = {
-        "Id": "e90e34656806",
+        "Id": "sha256:" + "a" * 64,
         "Warnings": []
     }
+    with asyncio_patch("gns3server.compute.docker.Docker.query", return_value=response) as mock:
+        vm = DockerVM("test", str(uuid.uuid4()), compute_project, manager, "ubuntu:latest",
+                      image_digest="sha256:" + "a" * 64)
+        await vm.create()
+        # the last query is the container creation: the digest check let it through
+        assert mock.call_args[0] == ("POST", "containers/create?name={}".format(vm.docker_name))
+        assert vm._cid == "sha256:" + "a" * 64
 
-    vm = DockerVM("test", str(uuid.uuid4()), compute_project, manager, "ubuntu")
-    vm._get_image_information = MagicMock()
-    vm._get_image_information.side_effect = information
-    with asyncio_patch("gns3server.compute.docker.DockerVM.pull_image", return_value=True) as mock_pull:
-        with asyncio_patch("gns3server.compute.docker.Docker.query", return_value=response) as mock:
+
+@pytest.mark.asyncio
+async def test_create_image_digest_mismatch(compute_project, manager):
+
+    response = {
+        "Id": "sha256:" + "b" * 64,
+        "Warnings": []
+    }
+    vm = DockerVM("test", str(uuid.uuid4()), compute_project, manager, "ubuntu:latest",
+                  image_digest="sha256:" + "a" * 64)
+    with asyncio_patch("gns3server.compute.docker.Docker.query", return_value=response) as query_mock:
+        with pytest.raises(ImageMissingError, match="ubuntu:latest"):
             await vm.create()
-            mock.assert_called_with("POST", "containers/create?name={}".format(vm.docker_name), data={
-                "Tty": True,
-                "OpenStdin": True,
-                "StdinOnce": False,
-                "HostConfig":
-                    {
-                        "CapAdd": ["ALL"],
-                        "Mounts": [
-                            {
-                                "Type": "bind",
-                                "Source": Docker.resources_path(),
-                                "Target": "/gns3",
-                                "ReadOnly": True
-                            },
-                            {
-                                "Type": "bind",
-                                "Source": os.path.join(vm.working_dir, "etc", "network"),
-                                "Target": "/gns3volumes/etc/network"
-                            }
-                        ],
-                        "Privileged": True,
-                        "Memory": 0,
-                        "NanoCpus": 0,
-                        "UsernsMode": "host"
-                    },
-                "Volumes": {},
-                "NetworkDisabled": True,
-                "Hostname": "test",
-                "Image": "ubuntu:latest",
-                "Env": [
-                    "container=docker",
-                    "GNS3_MAX_ETHERNET=eth0",
-                    "GNS3_VOLUMES=/etc/network"
-                    ],
-                "Entrypoint": ["/gns3/init.sh"],
-                "Cmd": ["/bin/sh"]
-            })
-        assert vm._cid == "e90e34656806"
-        mock_pull.assert_called_with("ubuntu:latest")
+        # only the image inspect happened: no container was created from the stale image
+        query_mock.assert_called_once_with("GET", "images/ubuntu:latest/json")
 
 
 @pytest.mark.asyncio

tests/controller/test_node.py

@@ -21,7 +21,8 @@ import uuid
 import os
 
 from unittest.mock import MagicMock, ANY
-from tests.utils import AsyncioMagicMock
+from tests.utils import AsyncioMagicMock, asyncio_patch
+from gns3server.compute.docker.docker_error import DockerError
 
 from gns3server.controller.node import Node
 from gns3server.controller.project import Project
@@ -693,6 +694,60 @@ async def test_upload_missing_image(compute, controller, images_dir):
     compute.post.assert_called_with("/qemu/images/linux.img", data=ANY, timeout=None)
 
 
+@pytest.mark.asyncio
+async def test_sync_missing_docker_image_from_controller_daemon(compute, controller):
+
+    project = Project(str(uuid.uuid4()), controller=controller)
+    node = Node(project, compute, "demo",
+                node_id=str(uuid.uuid4()),
+                node_type="docker",
+                properties={"image": "gns3/frr:latest"})
+
+    response = MagicMock()
+    response.content = MagicMock()
+    with asyncio_patch("gns3server.compute.docker.Docker.http_query", return_value=response) as save_mock:
+        assert await node._upload_missing_image("docker", "gns3/frr:latest") is True
+        save_mock.assert_called_with("GET", "images/gns3/frr:latest/get", timeout=None)
+    compute.post.assert_called_with("/docker/images/load", data=response.content, timeout=None)
+    response.close.assert_called_once_with()
+
+
+@pytest.mark.asyncio
+async def test_sync_missing_docker_image_pull_fallback(compute, controller):
+
+    project = Project(str(uuid.uuid4()), controller=controller)
+    node = Node(project, compute, "demo",
+                node_id=str(uuid.uuid4()),
+                node_type="docker",
+                properties={"image": "nginx:latest"})
+
+    with asyncio_patch("gns3server.compute.docker.Docker.http_query", side_effect=DockerError("404")):
+        assert await node._upload_missing_image("docker", "nginx:latest") is True
+    compute.post.assert_called_with("/docker/images/pull", data={"image": "nginx:latest"}, timeout=None)
+
+
+@pytest.mark.asyncio
+async def test_create_docker_node_pins_image_digest(compute, controller):
+
+    project = Project(str(uuid.uuid4()), controller=controller)
+    node = Node(project, compute, "demo",
+                node_id=str(uuid.uuid4()),
+                node_type="docker",
+                properties={"image": "gns3/frr:latest", "adapters": 1})
+
+    response = MagicMock()
+    response.status = 200
+    response.json = {}
+    compute.post = AsyncioMagicMock(return_value=response)
+
+    image_id = "sha256:" + "a" * 64
+    with asyncio_patch("gns3server.compute.docker.Docker.query", return_value={"Id": image_id}):
+        assert await node.create() is True
+    # the image id from the controller host daemon is pinned into the create payload
+    data = compute.post.call_args[1]["data"]
+    assert data["image_digest"] == image_id
+
+
 def test_update_label(node):
     """
     The text in label need to be always the