Managing Docker containers through Celery tasks.

config.ini

@@ -18,7 +18,7 @@ broker_url: redis://localhost:6379/1
 
 
 [PTChecker]
-jar_path: ~/JPTChecker-jar-with-dependencies.jar
+jar_path: /tmp/JPTChecker-jar-with-dependencies.jar
 
 
 [Port]

src/ptinstancemanager/tasks.py

@@ -1,10 +1,46 @@
 import logging
 from docker import Client
+from docker.errors import APIError
+from celery import chain
+
+import ptchecker
 from ptinstancemanager.app import app, celery
+from ptinstancemanager.models import Instance, Port, CachedFile
+
+
+
+def create_containers(num_containers):
+    logging.info('Creating new containers.')
+    for _ in range(num_containers):
+        available_port = Port.allocate()
+
+        if available_port is None:
+            raise Exception('The server cannot create new instances. Please, wait and retry it.')
+
+        res = create_container.apply_async((available_port.number,), link=wait_for_ready_container.s(10))
+	res.get()
 
 
 @celery.task()
-def create_container(pt_port, vnc_port):
+def create_container(pt_port):
+    logging.info('Creating new container.')
+
+    # Create container with Docker
+    vnc_port = pt_port + 10000
+    container_id = start_container(pt_port, vnc_port)
+
+    # If success...
+    instance = Instance.create(container_id, pt_port, vnc_port)
+    port = Port.get(pt_port)
+    port.assign(instance.id)
+
+    logging.info('Container started: %s' % container_id)
+
+    return instance.id
+
+
+#@celery.task()
+def start_container(pt_port, vnc_port):
     """Create container with Docker."""
     docker = Client(app.config['DOCKER_URL'], version='auto')
     port_bindings = { app.config['DOCKER_PT_PORT']: pt_port,
@@ -30,6 +66,54 @@ def create_container(pt_port, vnc_port):
 
 
 @celery.task()
-def stop_container(container_id):
+def stop_container(instance_id):
+    instance = Instance.get(instance_id)
+    docker = Client(app.config['DOCKER_URL'], version='auto')
+    try:
+        docker.stop(instance.docker_id)
+	instance.stop()
+	Port.get(instance.pt_port).release()  # The port can be now reused by a new PT instance
+    except APIError as ae:
+         # if it was already stopped or removed
+	if cli.containers(filters={'status': 'exited'}):  # Not empty array
+	    instance.stop()
+	    Port.get(instance.pt_port).release()  # The port can be now reused by a new PT instance
+
+
+@celery.task()
+def assign_container():
+    logging.info('Assigning container.')
+    docker = Client(app.config['DOCKER_URL'], version='auto')
+    for instance in Instance.get_unassigned():
+	try:
+	    docker.unpause(instance.docker_id)
+	    instance.assign()
+            return instance.id
+        except APIError as ae:
+	    # e.g., if it was already unpaused or it has been stopped
+	    instance.mark_error()
+	    logging.error('Docker API exception. %s.' % ae)
+
+
+@celery.task()
+def unassign_container(instance_id):
+    instance = Instance.get(instance_id)
     docker = Client(app.config['DOCKER_URL'], version='auto')
-    docker.stop(container_id)
+    try:
+        docker.pause(instance.docker_id)
+    except APIError as ae:
+	# e.g., if it was already paused
+	instance.mark_error()
+        logging.error('Docker API exception. %s.' % ae)
+    finally:
+        instance.unassign()
+
+
+@celery.task()
+def wait_for_ready_container(instance_id, timeout):
+    logging.info('Waiting for container to be ready.')
+    instance = Instance.get(instance_id)
+    is_running = ptchecker.is_running(app.config['PT_CHECKER'], 'localhost', instance.pt_port, float(timeout))
+    if not is_running:
+	raise Exception('The container has not answered yet.')
+    unassign_container.delay(instance_id)  # else