Merge pull request #379 from GNS3/docker_cleanup

Docker cleanup and improvements

gns3server/modules/__init__.py

@@ -27,6 +27,9 @@ MODULES = [VPCS, VirtualBox, Dynamips, Qemu, VMware]
 
 if sys.platform.startswith("linux") or hasattr(sys, "_called_from_test") or os.environ.get("PYTEST_BUILD_DOCUMENTATION") == "1":
 
+    from .docker import Docker
+    MODULES.append(Docker)
+
     # IOU runs only on Linux but testsuite work on UNIX platform
     if not sys.platform.startswith("win"):
         from .iou import IOU

gns3server/modules/adapters/ethernet_adapter.py

@@ -27,6 +27,7 @@ class EthernetAdapter(Adapter):
     def __init__(self, interfaces=1):
 
         super().__init__(interfaces)
+        self.host_ifc = None
 
     def __str__(self):
 

gns3server/modules/docker/__init__.py

@@ -22,51 +22,92 @@ Docker server module.
 import asyncio
 import logging
 import aiohttp
-import docker
-from requests.exceptions import ConnectionError
+import urllib
+import json
 
 log = logging.getLogger(__name__)
 
 from ..base_manager import BaseManager
 from ..project_manager import ProjectManager
-from .docker_vm import Container
+from .docker_vm import DockerVM
 from .docker_error import DockerError
 
 
 class Docker(BaseManager):
 
-    _VM_CLASS = Container
+    _VM_CLASS = DockerVM
 
     def __init__(self):
         super().__init__()
-        # FIXME: make configurable and start docker before trying
-        self._server_url = 'unix://var/run/docker.sock'
-        self._client = docker.Client(base_url=self._server_url)
-        self._execute_lock = asyncio.Lock()
-
-    @property
-    def server_url(self):
-        """Returns the Docker server url.
-
-        :returns: url
-        :rtype: string
-        """
-        return self._server_url
-
-    @server_url.setter
-    def server_url(self, value):
-        self._server_url = value
-        self._client = docker.Client(base_url=value)
+        self._server_url = '/var/run/docker.sock'
+        self._connector = aiohttp.connector.UnixConnector(self._server_url)
+        # Allow locking during ubridge operations
+        self.ubridge_lock = asyncio.Lock()
 
     @asyncio.coroutine
-    def execute(self, command, kwargs, timeout=60):
-        command = getattr(self._client, command)
-        log.debug("Executing Docker with command: {}".format(command))
-        try:
-            result = command(**kwargs)
-        except Exception as error:
-            raise DockerError("Docker has returned an error: {}".format(error))
-        return result
+    def query(self, method, path, data={}, params={}):
+        """
+        Make a query to the docker daemon and decode the request
+
+        :param method: HTTP method
+        :param path: Endpoint in API
+        :param data: Dictionnary with the body. Will be transformed to a JSON
+        :param params: Parameters added as a query arg
+        """
+        response = yield from self.http_query(method, path, data=data, params=params)
+        body = yield from response.read()
+        if len(body):
+            body = json.loads(body.decode("utf-8"))
+        log.debug("Query Docker %s %s params=%s data=%s Response: %s", method, path, params, data, body)
+        return body
+
+    @asyncio.coroutine
+    def http_query(self, method, path, data={}, params={}):
+        """
+        Make a query to the docker daemon
+
+        :param method: HTTP method
+        :param path: Endpoint in API
+        :param data: Dictionnary with the body. Will be transformed to a JSON
+        :param params: Parameters added as a query arg
+        :returns: HTTP response
+        """
+        data = json.dumps(data)
+        url = "http://docker/" + path
+        response = yield from aiohttp.request(
+            method,
+            url,
+            connector=self._connector,
+            params=params,
+            data=data,
+            headers={"content-type": "application/json", },
+        )
+        if response.status >= 300:
+            body = yield from response.read()
+            try:
+                body = json.loads(body.decode("utf-8"))["message"]
+            except ValueError:
+                pass
+            log.debug("Query Docker %s %s params=%s data=%s Response: %s", method, path, params, data, body)
+            raise DockerError("Docker has returned an error: {}".format(body))
+        return response
+
+    @asyncio.coroutine
+    def websocket_query(self, path, params={}):
+        """
+        Open a websocket connection
+
+        :param path: Endpoint in API
+        :param params: Parameters added as a query arg
+        :returns: Websocket
+        """
+
+        url = "http://docker/" + path
+        connection = yield from aiohttp.ws_connect(url,
+                                                   connector=self._connector,
+                                                   origin="http://docker",
+                                                   autoping=True)
+        return connection
 
     @asyncio.coroutine
     def list_images(self):
@@ -76,44 +117,8 @@ class Docker(BaseManager):
         :rtype: list
         """
         images = []
-        try:
-            for image in self._client.images():
-                for tag in image['RepoTags']:
-                    images.append({'imagename': tag})
-            return images
-        except ConnectionError as error:
-            raise DockerError(
-                """Docker couldn't list images and returned an error: {}
-Is the Docker service running?""".format(error))
-
-    @asyncio.coroutine
-    def list_containers(self):
-        """Gets Docker container list.
-
-        :returns: list of dicts
-        :rtype: list
-        """
-        return self._client.containers()
-
-    def get_container(self, cid, project_id=None):
-        """Returns a Docker container.
-
-        :param id: Docker container identifier
-        :param project_id: Project identifier
-
-        :returns: Docker container
-        """
-        if project_id:
-            project = ProjectManager.instance().get_project(project_id)
-
-        if cid not in self._vms:
-            raise aiohttp.web.HTTPNotFound(
-                text="Docker container with ID {} doesn't exist".format(cid))
-
-        container = self._vms[cid]
-        if project_id:
-            if container.project.id != project.id:
-                raise aiohttp.web.HTTPNotFound(
-                    text="Project ID {} doesn't belong to container {}".format(
-                        project_id, container.name))
-        return container
+        for image in (yield from self.query("GET", "images/json", params={"all": 0})):
+            for tag in image['RepoTags']:
+                if tag != "<none>:<none>":
+                    images.append({'image': tag})
+        return sorted(images, key=lambda i: i['image'])

gns3server/modules/docker/docker_vm.py

@@ -22,20 +22,22 @@ Docker container instance.
 import asyncio
 import shutil
 import psutil
+import shlex
+import aiohttp
+import json
 
-from docker.utils import create_host_config
-from gns3server.ubridge.hypervisor import Hypervisor
-from pkg_resources import parse_version
+from ...ubridge.hypervisor import Hypervisor
 from .docker_error import DockerError
 from ..base_vm import BaseVM
 from ..adapters.ethernet_adapter import EthernetAdapter
 from ..nios.nio_udp import NIOUDP
+from ...utils.asyncio.telnet_server import AsyncioTelnetServer
 
 import logging
 log = logging.getLogger(__name__)
 
 
-class Container(BaseVM):
+class DockerVM(BaseVM):
     """Docker container implementation.
 
     :param name: Docker container name
@@ -45,18 +47,23 @@ class Container(BaseVM):
     :param image: Docker image
     """
 
-    def __init__(self, name, vm_id, project, manager, image, startcmd=None):
-        self._name = name
-        self._id = vm_id
-        self._project = project
-        self._manager = manager
+    def __init__(self, name, vm_id, project, manager, image, console=None, start_command=None, adapters=None, environment=None):
+        super().__init__(name, vm_id, project, manager, console=console)
+
         self._image = image
-        self._startcmd = startcmd
-        self._veths = []
+        self._start_command = start_command
+        self._environment = environment
+        self._cid = None
         self._ethernet_adapters = []
         self._ubridge_hypervisor = None
         self._temporary_directory = None
-        self._hw_virtualization = False
+        self._telnet_server = None
+        self._closed = False
+
+        if adapters is None:
+            self.adapters = 1
+        else:
+            self.adapters = adapters
 
         log.debug(
             "{module}: {name} [{image}] initialized.".format(
@@ -68,15 +75,30 @@ class Container(BaseVM):
         return {
             "name": self._name,
             "vm_id": self._id,
-            "cid": self._cid,
+            "container_id": self._cid,
             "project_id": self._project.id,
             "image": self._image,
+            "adapters": self.adapters,
+            "console": self.console,
+            "start_command": self.start_command,
+            "environment": self.environment
         }
 
     @property
-    def veths(self):
-        """Returns Docker host veth interfaces."""
-        return self._veths
+    def start_command(self):
+        return self._start_command
+
+    @start_command.setter
+    def start_command(self, command):
+        self._start_command = command
+
+    @property
+    def environment(self):
+        return self._environment
+
+    @environment.setter
+    def environment(self, command):
+        self._environment = command
 
     @asyncio.coroutine
     def _get_container_state(self):
@@ -85,74 +107,53 @@ class Container(BaseVM):
         :returns: state
         :rtype: str
         """
-        try:
-            result = yield from self.manager.execute(
-                "inspect_container", {"container": self._cid})
-            result_dict = {state.lower(): value for state, value in result["State"].items()}
-            for state, value in result_dict.items():
-                if value is True:
-                    # a container can be both paused and running
-                    if state == "paused":
-                        return "paused"
-                    if state == "running":
-                        if "paused" in result_dict and result_dict["paused"] is True:
-                            return "paused"
-                    return state.lower()
-            return 'exited'
-        except Exception as err:
-            raise DockerError("Could not get container state for {0}: ".format(
-                self._name), str(err))
+        result = yield from self.manager.query("GET", "containers/{}/json".format(self._cid))
+
+        if result["State"]["Paused"]:
+            return "paused"
+        if result["State"]["Running"]:
+            return "running"
+        return "exited"
 
     @asyncio.coroutine
     def create(self):
         """Creates the Docker container."""
         params = {
-            "name": self._name,
-            "image": self._image,
-            "network_disabled": True,
-            "host_config": create_host_config(
-                privileged=True, cap_add=['ALL'])
+            "Name": self._name,
+            "Image": self._image,
+            "NetworkDisabled": True,
+            "Tty": True,
+            "OpenStdin": True,
+            "StdinOnce": False,
+            "HostConfig": {
+                "CapAdd": ["ALL"],
+                "Privileged": True
+            }
         }
-        if self._startcmd:
-            params.update({'command': self._startcmd})
+        if self._start_command:
+            params.update({"Cmd": shlex.split(self._start_command)})
 
-        result = yield from self.manager.execute("create_container", params)
+        if self._environment:
+            params.update({"Env": [e.strip() for e in self._environment.split("\n")]})
+
+        images = [i["image"] for i in (yield from self.manager.list_images())]
+        if self._image not in images:
+            log.info("Image %s is missing pulling it from docker hub", self._image)
+            yield from self.pull_image(self._image)
+
+        result = yield from self.manager.query("POST", "containers/create", data=params)
         self._cid = result['Id']
         log.info("Docker container '{name}' [{id}] created".format(
             name=self._name, id=self._id))
         return True
 
-    @property
-    def ubridge_path(self):
-        """Returns the uBridge executable path.
-
-        :returns: path to uBridge
-        """
-        path = self._manager.config.get_section_config("Server").get(
-            "ubridge_path", "ubridge")
-        if path == "ubridge":
-            path = shutil.which("ubridge")
-        return path
-
     @asyncio.coroutine
-    def _start_ubridge(self):
-        """Starts uBridge (handles connections to and from this Docker VM)."""
-        server_config = self._manager.config.get_section_config("Server")
-        server_host = server_config.get("host")
-        self._ubridge_hypervisor = Hypervisor(
-            self._project, self.ubridge_path, self.working_dir, server_host)
-
-        log.info("Starting new uBridge hypervisor {}:{}".format(
-            self._ubridge_hypervisor.host, self._ubridge_hypervisor.port))
-        yield from self._ubridge_hypervisor.start()
-        log.info("Hypervisor {}:{} has successfully started".format(
-            self._ubridge_hypervisor.host, self._ubridge_hypervisor.port))
-        yield from self._ubridge_hypervisor.connect()
-        if parse_version(
-                self._ubridge_hypervisor.version) < parse_version('0.9.1'):
-            raise DockerError(
-                "uBridge version must be >= 0.9.1, detected version is {}".format(
-                    self._ubridge_hypervisor.version))
+    def update(self):
+        """
+        Destroy an recreate the container with the new settings
+        """
+        yield from self.remove()
+        yield from self.create()
 
     @asyncio.coroutine
     def start(self):
@@ -162,17 +163,68 @@ class Container(BaseVM):
         if state == "paused":
             yield from self.unpause()
         else:
-            result = yield from self.manager.execute(
-                "start", {"container": self._cid})
+            result = yield from self.manager.query("POST", "containers/{}/start".format(self._cid))
 
-        yield from self._start_ubridge()
-        for adapter_number in range(0, self.adapters):
-            nio = self._ethernet_adapters[adapter_number].get_nio(0)
-            if nio:
-                yield from self._add_ubridge_connection(nio, adapter_number)
+            yield from self._start_ubridge()
+            for adapter_number in range(0, self.adapters):
+                nio = self._ethernet_adapters[adapter_number].get_nio(0)
+                if nio:
+                    with (yield from self.manager.ubridge_lock):
+                        yield from self._add_ubridge_connection(nio, adapter_number)
 
-        log.info("Docker container '{name}' [{image}] started".format(
-            name=self._name, image=self._image))
+            yield from self._start_console()
+
+        self.status = "started"
+        log.info("Docker container '{name}' [{image}] started listen for telnet on {console}".format(name=self._name, image=self._image, console=self._console))
+
+    @asyncio.coroutine
+    def _start_console(self):
+        """
+        Start streaming the console via telnet
+        """
+        class InputStream:
+
+            def __init__(self):
+                self._data = b""
+
+            def write(self, data):
+                self._data += data
+
+            @asyncio.coroutine
+            def drain(self):
+                if not self.ws.closed:
+                    self.ws.send_bytes(self._data)
+                self._data = b""
+
+        output_stream = asyncio.StreamReader()
+        input_stream = InputStream()
+
+        telnet = AsyncioTelnetServer(reader=output_stream, writer=input_stream)
+        self._telnet_server = yield from asyncio.start_server(telnet.run, self._manager.port_manager.console_host, self._console)
+
+        ws = yield from self.manager.websocket_query("containers/{}/attach/ws?stream=1&stdin=1&stdout=1&stderr=1".format(self._cid))
+        input_stream.ws = ws
+
+        output_stream.feed_data(self.name.encode() + b" console is now available... Press RETURN to get started.\r\n")
+
+        asyncio.async(self._read_console_output(ws, output_stream))
+
+    @asyncio.coroutine
+    def _read_console_output(self, ws, out):
+        """
+        Read websocket and forward it to the telnet
+        :params ws: Websocket connection
+        :param out: Output stream
+        """
+
+        while True:
+            msg = yield from ws.receive()
+            if msg.tp == aiohttp.MsgType.text:
+                out.feed_data(msg.data.encode())
+            else:
+                out.feed_eof()
+                ws.close()
+                break
 
     def is_running(self):
         """Checks if the container is running.
@@ -187,9 +239,8 @@ class Container(BaseVM):
 
     @asyncio.coroutine
     def restart(self):
-        """Restarts this Docker container."""
-        result = yield from self.manager.execute(
-            "restart", {"container": self._cid})
+        """Restart this Docker container."""
+        yield from self.manager.query("POST", "containers/{}/restart".format(self._cid))
         log.info("Docker container '{name}' [{image}] restarted".format(
             name=self._name, image=self._image))
 
@@ -203,27 +254,30 @@ class Container(BaseVM):
         state = yield from self._get_container_state()
         if state == "paused":
             yield from self.unpause()
-        result = yield from self.manager.execute(
-            "kill", {"container": self._cid})
+
+        if self._telnet_server:
+            self._telnet_server.close()
+            self._telnet_server = None
+        # t=5 number of seconds to wait before killing the container
+        yield from self.manager.query("POST", "containers/{}/stop".format(self._cid), params={"t": 5})
         log.info("Docker container '{name}' [{image}] stopped".format(
             name=self._name, image=self._image))
 
     @asyncio.coroutine
     def pause(self):
         """Pauses this Docker container."""
-        result = yield from self.manager.execute(
-            "pause", {"container": self._cid})
+        yield from self.manager.query("POST", "containers/{}/pause".format(self._cid))
         log.info("Docker container '{name}' [{image}] paused".format(
             name=self._name, image=self._image))
+        self.status = "paused"
 
     @asyncio.coroutine
     def unpause(self):
         """Unpauses this Docker container."""
-        result = yield from self.manager.execute(
-            "unpause", {"container": self._cid})
-        state = yield from self._get_container_state()
+        yield from self.manager.query("POST", "containers/{}/unpause".format(self._cid))
         log.info("Docker container '{name}' [{image}] unpaused".format(
             name=self._name, image=self._image))
+        self.status = "started"
 
     @asyncio.coroutine
     def remove(self):
@@ -233,17 +287,30 @@ class Container(BaseVM):
             yield from self.unpause()
         if state == "running":
             yield from self.stop()
-        result = yield from self.manager.execute(
-            "remove_container", {"container": self._cid, "force": True})
+        yield from self.manager.query("DELETE", "containers/{}".format(self._cid), params={"force": 1})
         log.info("Docker container '{name}' [{image}] removed".format(
             name=self._name, image=self._image))
 
+        if self._console:
+            self._manager.port_manager.release_tcp_port(self._console, self._project)
+            self._console = None
+
+        for adapter in self._ethernet_adapters:
+            if adapter is not None:
+                for nio in adapter.ports.values():
+                    if nio and isinstance(nio, NIOUDP):
+                        self.manager.port_manager.release_udp_port(nio.lport, self._project)
+
     @asyncio.coroutine
     def close(self):
         """Closes this Docker container."""
+
+        if self._closed:
+            return
+
         log.debug("Docker container '{name}' [{id}] is closing".format(
             name=self.name, id=self._cid))
-        for adapter in self._ethernet_adapters.values():
+        for adapter in self._ethernet_adapters:
             if adapter is not None:
                 for nio in adapter.ports.values():
                     if nio and isinstance(nio, NIOUDP):
@@ -256,6 +323,7 @@ class Container(BaseVM):
             name=self.name, id=self._cid))
         self._closed = True
 
+    @asyncio.coroutine
     def _add_ubridge_connection(self, nio, adapter_number):
         """
         Creates a connection in uBridge.
@@ -267,8 +335,7 @@ class Container(BaseVM):
             adapter = self._ethernet_adapters[adapter_number]
         except IndexError:
             raise DockerError(
-                "Adapter {adapter_number} doesn't exist on Docker container '{name}'".format(
-                    name=self.name, adapter_number=adapter_number))
+                "Adapter {adapter_number} doesn't exist on Docker container '{name}'".format(name=self.name, adapter_number=adapter_number))
 
         if nio and isinstance(nio, NIOUDP):
             for index in range(128):
@@ -279,18 +346,20 @@ class Container(BaseVM):
                     break
             if not hasattr(adapter, "ifc"):
                 raise DockerError(
-                    "Adapter {adapter_number} couldn't allocate interface on Docker container '{name}'".format(
+                    "Adapter {adapter_number} couldn't allocate interface on Docker container '{name}'. Too many Docker interfaces already exists".format(
                         name=self.name, adapter_number=adapter_number))
+        else:
+            raise ValueError("Invalid NIO")
 
         yield from self._ubridge_hypervisor.send(
             'docker create_veth {hostif} {guestif}'.format(
                 guestif=adapter.guest_ifc, hostif=adapter.host_ifc))
-        self._veths.append(adapter.host_ifc)
 
-        namespace = yield from self.get_namespace()
+        namespace = yield from self._get_namespace()
+        log.debug("Move container %s adapter %s to namespace %s", self.name, adapter.guest_ifc, namespace)
         yield from self._ubridge_hypervisor.send(
-            'docker move_to_ns {ifc} {ns}'.format(
-                ifc=adapter.guest_ifc, ns=namespace))
+            'docker move_to_ns {ifc} {ns} eth{adapter}'.format(
+                ifc=adapter.guest_ifc, ns=namespace, adapter=adapter_number))
 
         yield from self._ubridge_hypervisor.send(
             'bridge create bridge{}'.format(adapter_number))
@@ -321,9 +390,14 @@ class Container(BaseVM):
             name=adapter_number))
 
         adapter = self._ethernet_adapters[adapter_number]
-        yield from self._ubridge_hypervisor.send("docker delete_veth {name}".format(
-            name=adapter.host_ifc))
+        yield from self._ubridge_hypervisor.send('docker delete_veth {hostif} {guestif}'.format(guestif=adapter.guest_ifc, hostif=adapter.host_ifc))
 
+    @asyncio.coroutine
+    def _get_namespace(self):
+        result = yield from self.manager.query("GET", "containers/{}/json".format(self._cid))
+        return int(result['State']['Pid'])
+
+    @asyncio.coroutine
     def adapter_add_nio_binding(self, adapter_number, nio):
         """Adds an adapter NIO binding.
 
@@ -345,6 +419,7 @@ class Container(BaseVM):
                 nio=nio,
                 adapter_number=adapter_number))
 
+    @asyncio.coroutine
     def adapter_remove_nio_binding(self, adapter_number):
         """
         Removes an adapter NIO binding.
@@ -361,10 +436,7 @@ class Container(BaseVM):
                     name=self.name, adapter_number=adapter_number))
 
         adapter.remove_nio(0)
-        try:
-            yield from self._delete_ubridge_connection(adapter_number)
-        except:
-            pass
+        yield from self._delete_ubridge_connection(adapter_number)
 
         log.info(
             "Docker VM '{name}' [{id}]: {nio} removed from adapter {adapter_number}".format(
@@ -397,7 +469,28 @@ class Container(BaseVM):
                 id=self._id,
                 adapters=adapters))
 
-    def get_namespace(self):
-        result = yield from self.manager.execute(
-            "inspect_container", {"container": self._cid})
-        return int(result['State']['Pid'])
+    @asyncio.coroutine
+    def pull_image(self, image):
+        """
+        Pull image from docker repository
+        """
+        log.info("Pull %s from docker hub", image)
+        response = yield from self.manager.http_query("POST", "images/create", params={"fromImage": image})
+        # The pull api will stream status via an HTTP JSON stream
+        content = ""
+        while True:
+            chunk = yield from response.content.read(1024)
+            if not chunk:
+                break
+            content += chunk.decode("utf-8")
+
+            try:
+                while True:
+                    content = content.lstrip(" \r\n\t")
+                    answer, index = json.JSONDecoder().raw_decode(content)
+                    if "progress" in answer:
+                        self.project.emit("log.info", {"message": "Pulling image {}:{}: {}".format(self._image, answer["id"], answer["progress"])})
+                    content = content[index:]
+            except ValueError:  # Partial JSON
+                pass
+        self.project.emit("log.info", {"message": "Success pulling image {}".format(self._image)})