diff --git a/gns3server/api/routes/compute/docker_nodes.py b/gns3server/api/routes/compute/docker_nodes.py index 4c7f319c3..a5e48c57d 100644 --- a/gns3server/api/routes/compute/docker_nodes.py +++ b/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(): diff --git a/gns3server/api/routes/compute/images.py b/gns3server/api/routes/compute/images.py index 2bde73906..fd49974a1 100644 --- a/gns3server/api/routes/compute/images.py +++ b/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]: """ diff --git a/gns3server/compute/docker/__init__.py b/gns3server/compute/docker/__init__.py index f4027ea59..39ca9f20a 100644 --- a/gns3server/compute/docker/__init__.py +++ b/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. diff --git a/gns3server/compute/docker/docker_vm.py b/gns3server/compute/docker/docker_vm.py index 7f4812e7b..bfcc53a7e 100644 --- a/gns3server/compute/docker/docker_vm.py +++ b/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. diff --git a/gns3server/controller/node.py b/gns3server/controller/node.py index c9189f8d6..452dfb984 100644 --- a/gns3server/controller/node.py +++ b/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 diff --git a/gns3server/schemas/compute/docker_nodes.py b/gns3server/schemas/compute/docker_nodes.py index e2023d407..d752b7586 100644 --- a/gns3server/schemas/compute/docker_nodes.py +++ b/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:, 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): diff --git a/tests/api/routes/compute/test_images.py b/tests/api/routes/compute/test_images.py index d4331557c..9aff40a3d 100644 --- a/tests/api/routes/compute/test_images.py +++ b/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 diff --git a/tests/compute/docker/test_docker.py b/tests/compute/docker/test_docker.py index 7fbb49c06..c00d6c11b 100644 --- a/tests/compute/docker/test_docker.py +++ b/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): diff --git a/tests/compute/docker/test_docker_vm.py b/tests/compute/docker/test_docker_vm.py index 92e790913..51359b024 100644 --- a/tests/compute/docker/test_docker_vm.py +++ b/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 diff --git a/tests/controller/test_node.py b/tests/controller/test_node.py index e4ed15bd6..b62784052 100644 --- a/tests/controller/test_node.py +++ b/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