From ec2b098c29a711ba48d4ba5a8075c839d66085b9 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?J=C3=A9r=C3=B4me=20Petazzoni?= <jerome.petazzoni@gmail.com>
Date: Mon, 29 Feb 2016 21:24:32 +0000
Subject: [PATCH] Add parallel_run function and use it to manage ambassadors

---
 bin/common.py                | 40 ++++++++++++++++++++++++++++++++++++
 bin/configure-ambassadors.py | 17 +++++++++------
 bin/create-ambassadors.py    | 18 +++++++++-------
 3 files changed, 62 insertions(+), 13 deletions(-)

diff --git a/bin/common.py b/bin/common.py
index 0416b964..3d2f83f2 100755
--- a/bin/common.py
+++ b/bin/common.py
@@ -1,4 +1,5 @@
 import os
+import subprocess
 import sys
 import time
 import yaml
@@ -34,3 +35,42 @@ class ComposeFile(object):
         with open(filename, "w") as f:
             yaml.safe_dump(self.data, f, default_flow_style=False)
 
+# Executes a bunch of commands in parallel, but no more than N at a time.
+# This allows to execute concurrently a large number of tasks, without
+# turning into a fork bomb.
+# `parallelism` is the number of tasks to execute simultaneously.
+# `commands` is a list of tasks to execute.
+# Each task is itself a list, where the first element is a descriptive
+# string, and the folloowing elements are the arguments to pass to Popen.
+def parallel_run(commands, parallelism):
+    running = []
+    # While stuff is running, or we have stuff to run...
+    while commands or running:
+        # While there is stuff to run, and room in the pipe...
+        while commands and len(running)<parallelism:
+            command = commands.pop(0)
+            print("START {}".format(command[0]))
+            popen = subprocess.Popen(
+                command[1:], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
+            popen._desc = command[0]
+            running.append(popen)
+        must_sleep = True
+        for popen in running:
+            status = popen.poll()
+            if status is not None:
+                must_sleep = False
+                running.remove(popen)
+                if status==0:
+                    print("OK    {}".format(popen._desc))
+                else:
+                    print("ERROR {} [Exit status: {}]"
+                          .format(popen._desc, status))
+                    output = "\n" + popen.communicate()[0].strip()
+                    output = output.replace("\n", "\n| ")
+                    print(output)
+        else:
+            print("WAIT  ({} running, {} more to run)"
+                  .format(len(running), len(commands)))
+            if must_sleep:
+                time.sleep(1)
+
diff --git a/bin/configure-ambassadors.py b/bin/configure-ambassadors.py
index d226fec6..06fec8d1 100755
--- a/bin/configure-ambassadors.py
+++ b/bin/configure-ambassadors.py
@@ -1,11 +1,12 @@
 #!/usr/bin/env python
 
+from common import parallel_run
 import os
 import subprocess
 
 project_name = os.path.basename(os.path.realpath("."))
 
-# Get all services and backends in our compose application
+# Get all services and backends in our compose application.
 containers_data = subprocess.check_output([
     "docker", "ps",
     "--filter", "label=com.docker.compose.project={}".format(project_name),
@@ -14,7 +15,7 @@ containers_data = subprocess.check_output([
                 '{{ .Ports }}',
 ])
 
-# Build list of backends
+# Build list of backends.
 frontend_ports = dict()
 backends = dict()
 for container in containers_data.split('\n'):
@@ -35,7 +36,7 @@ for container in containers_data.split('\n'):
         backends[service_name] = []
     backends[service_name].append((backend_addr, backend_port))
 
-# Get all existing ambassadors for this application
+# Get all existing ambassadors for this application.
 ambassadors_data = subprocess.check_output([
     "docker", "ps",
     "--filter", "label=ambassador.project={}".format(project_name),
@@ -44,7 +45,8 @@ ambassadors_data = subprocess.check_output([
                 '{{ .Label "ambassador.bindaddr" }}',
 ])
 
-# Update ambassadors
+# Update ambassadors.
+operations = []
 for ambassador in ambassadors_data.split('\n'):
     if not ambassador:
         continue
@@ -54,11 +56,14 @@ for ambassador in ambassadors_data.split('\n'):
                   bind_address, frontend_ports[service_name],
                   backends[service_name]))
     command = [
+        ambassador_id,
         "docker", "run", "--rm", "--volumes-from", ambassador_id,
         "jpetazzo/hamba", "reconfigure",
         "{}:{}".format(bind_address, frontend_ports[service_name])
     ]
     for backend_addr, backend_port in backends[service_name]:
         command.extend([backend_addr, backend_port])
-    print command
-    subprocess.check_call(command)
+    operations.append(command)
+
+# Execute all commands in parallel.
+parallel_run(operations, 10)
diff --git a/bin/create-ambassadors.py b/bin/create-ambassadors.py
index 1e34fa93..92705f0a 100755
--- a/bin/create-ambassadors.py
+++ b/bin/create-ambassadors.py
@@ -1,6 +1,6 @@
 #!/usr/bin/env python
 
-from common import ComposeFile
+from common import ComposeFile, parallel_run
 import os
 import subprocess
 
@@ -8,14 +8,14 @@ config = ComposeFile()
 
 project_name = os.path.basename(os.path.realpath("."))
 
-# Get all services in our compose application
+# Get all services in our compose application.
 containers_data = subprocess.check_output([
     "docker", "ps",
     "--filter", "label=com.docker.compose.project={}".format(project_name),
     "--format", '{{ .ID }} {{ .Label "com.docker.compose.service" }}',
 ])
 
-# Get all existing ambassadors for this application
+# Get all existing ambassadors for this application.
 ambassadors_data = subprocess.check_output([
     "docker", "ps",
     "--filter", "label=ambassador.project={}".format(project_name),
@@ -24,7 +24,7 @@ ambassadors_data = subprocess.check_output([
                 '{{ .Label "ambassador.service" }}',
 ])
 
-# Build a set of existing ambassadors
+# Build a set of existing ambassadors.
 ambassadors = dict()
 for ambassador in ambassadors_data.split('\n'):
     if not ambassador:
@@ -32,7 +32,8 @@ for ambassador in ambassadors_data.split('\n'):
     ambassador_id, container_id, linked_service = ambassador.split()
     ambassadors[container_id, linked_service] = ambassador_id
 
-# Start the missing ambassadors
+# Start the missing ambassadors.
+operations = []
 for container in containers_data.split('\n'):
     if not container:
         continue
@@ -45,8 +46,9 @@ for container in containers_data.split('\n'):
         if ambassador_id:
             print("{} already exists: {}".format(description, ambassador_id))
         else:
-            print("{} not found, creating it:".format(description))
-	    subprocess.check_call([
+            print("{} not found, creating it.".format(description))
+	    operations.append([
+                description,
 		"docker", "run", "-d",
 		"--net", "container:{}".format(container_id),
 		"--label", "ambassador.project={}".format(project_name),
@@ -56,3 +58,5 @@ for container in containers_data.split('\n'):
 		"jpetazzo/hamba", "run"
 	    ])
 
+# Execute all commands in parallel.
+parallel_run(operations, 10)