Repository: OpenMined/PySyft
Pull request: Split Docker Launch into Pull / Build and Start
## Description
Please include a summary of the change, the motivation, and any additional context that will help others understand your PR. If it closes one or more open issues, [please tag them as described here](https://help.github.com/en/github/managing-your-work-on-github/linking-a-pull-request-to-an-issue#linking-a-pull-request-to-an-issue-using-a-keyword).

## Affected Dependencies
List any dependencies that are required for this change.

## How has this been tested?
- Describe the tests that you ran to verify your changes.
- Provide instructions so we can reproduce.
- List any relevant details for your test configuration.

## Checklist
- [ ] I have followed the [Contribution Guidelines](https://github.com/OpenMined/.github/blob/master/CONTRIBUTING.md) and [Code of Conduct](https://github.com/OpenMined/.github/blob/master/CODE_OF_CONDUCT.md)
- [ ] I have commented my code following the [OpenMined Styleguide](https://github.com/OpenMined/.github/blob/master/STYLEGUIDE.md)
- [ ] I have labeled this PR with the relevant [Type labels](https://github.com/OpenMined/.github/labels?q=Type%3A)
- [ ] My changes are covered by tests


Problem statement:
Split Docker Launch into Pull / Build and Start
To provide more control over the output filtering and checking what step is working we should split docker launch into docker pull (if they use tag=latest etc) or docker build (if they are building), run those commands and then after they are complete run the next one to launch the domain.

During the output of `docker compose pull` or `docker compose build` we can try to filter the stdout / stderr using the non blocking threaded subprocess example we now use in `hagrid` cli.py to catch the jupyter url in the output, look for:

```
queue: Queue = Queue()
thread_1 = Thread(target=enqueue_output, args=(proc.stdout, queue))
thread_2 = Thread(target=enqueue_output, args=(proc.stderr, queue))
```

As these steps are completed lets output something like:
✅  pulling containers
✅  building containers
✅  deploying containers
✅  waiting for health checks

If possible some small animation or dots or progress would be good just to keep users engaged.

Then we can remove the print "success" which is unhelpful.

We can add the --check=True as a default arg so that we automatically verify that docker launches correctly as part of the launch command (unless turned off)

Once this works we can remove `hagrid --silent` from our example notebooks so that we always give output but dont break the notebook with the zillion lines of animation spam docker outputs. We can also remove the `manual check example`

Hints from issue discussion:
Very much looking forward to this one!
@factdroid can we up-prioritize this one? Looks like it's been chillin for a month or so.



> @factdroid can we up-prioritize this one? Looks like it's been chillin for a month or so.

@iamtrask yeah it's a priority for me this week, I'll try to push through it.

Commit message: Fix Building step tracking issue
- Now we have a reliable method that will count how much images
we're trying to build based on matching regex pattern
- Now we have a reliable method that will count how much images
we built base on match regex pattern

Patch to review:
diff --git a/packages/grid/.env b/packages/grid/.env
index 79c9dd55249..207f658f495 100644
--- a/packages/grid/.env
+++ b/packages/grid/.env
@@ -25,6 +25,7 @@ DOCKER_IMAGE_VPN_IPTABLES=openmined/grid-vpn-iptables
 DOCKER_IMAGE_TRAEFIK=traefik
 TRAEFIK_VERSION=v2.8.1
 REDIS_VERSION=6.2
+RABBITMQ_VERSION=3
 DOCKER_IMAGE_SEAWEEDFS=chrislusf/seaweedfs:latest
 VERSION=latest
 VERSION_HASH=unknown
diff --git a/packages/grid/docker-compose.pull.yml b/packages/grid/docker-compose.pull.yml
new file mode 100644
index 00000000000..cf6df982405
--- /dev/null
+++ b/packages/grid/docker-compose.pull.yml
@@ -0,0 +1,19 @@
+version: "3.8"
+services:
+  db:
+    image: postgres:12
+
+  redis:
+    image: redis:${REDIS_VERSION?Variable not set}
+
+  queue:
+    image: rabbitmq:${RABBITMQ_VERSION?Variable not Set}${RABBITMQ_MANAGEMENT:-}
+
+  seaweedfs:
+    image: "${DOCKER_IMAGE_SEAWEEDFS?Variable not set}"
+
+  docker-host:
+    image: qoomon/docker-host
+
+  proxy:
+    image: ${DOCKER_IMAGE_TRAEFIK?Variable not set}:${TRAEFIK_VERSION?Variable not set}
diff --git a/packages/hagrid/hagrid/cli.py b/packages/hagrid/hagrid/cli.py
index 95bdfbaab3a..da3228e426b 100644
--- a/packages/hagrid/hagrid/cli.py
+++ b/packages/hagrid/hagrid/cli.py
@@ -5,6 +5,7 @@
 from queue import Queue
 import re
 import shutil
+import signal as sys_signal
 import socket
 import stat
 import subprocess  # nosec
@@ -31,6 +32,7 @@
 import rich
 from rich.console import Console
 from rich.live import Live
+from rich.progress import Progress
 from virtualenvapi.manage import VirtualEnvironment
 
 # relative
@@ -343,7 +345,9 @@ def launch(args: TypeTuple[str], **kwargs: Any) -> None:
         dry_run = False
 
     try:
-        silent = bool(kwargs["silent"]) if "silent" in kwargs else False
+        tail = False if "tail" in kwargs and not str_to_bool(kwargs["tail"]) else True
+        silent = not tail
+
         from_rendered_dir = (
             str_to_bool(cast(str, kwargs["from_template"])) and EDITABLE_MODE
         )
@@ -351,40 +355,174 @@ def launch(args: TypeTuple[str], **kwargs: Any) -> None:
         execute_commands(
             cmds, dry_run=dry_run, silent=silent, from_rendered_dir=from_rendered_dir
         )
-        print("Success!\n\n")
+        host_term = verb.get_named_term_hostgrammar(name="host")
+        command = cmds[list(cmds.keys())[0]][0]  # type: ignore
+        match_port = re.search("HTTP_PORT=[0-9]{1,5}", command)
+
+        if host_term.host == "docker" and match_port and silent:
+            rich.get_console().print(
+                "\n[bold green]⠋[bold blue] Checking  Node API [/bold blue]\t"
+            )
+            port = match_port.group().replace("HTTP_PORT=", "")
+            check_status("localhost" + ":" + port)
     except Exception as e:
         print(f"Error: {e}\n\n")
         return
 
 
-def execute_commands(
-    cmds: TypeList,
-    dry_run: bool = False,
-    silent: bool = False,
-    from_rendered_dir: bool = False,
-) -> None:
-    """Execute the launch commands and display their status in realtime.
+def check_errors(line: str, pid: int, cmd_name: str, progress_bar: Progress) -> None:
+    task = progress_bar.tasks[0]
+    if "Error response from daemon: " in line:
+        if progress_bar:
+            progress_bar.update(
+                0,
+                description=f"❌ [bold red]{cmd_name}[/bold red] [{task.completed} / {task.total}]",
+                refresh=True,
+            )
+            progress_bar.update(0, visible=False)
+            progress_bar.console.clear_live()
+            progress_bar.console.quiet = True
+            progress_bar.stop()
+            console = rich.get_console()
+            progress_bar.console.quiet = False
+            console.print(f"\n\n [red] ERROR [/red]: [bold]{line}[/bold]\n")
+        os.killpg(os.getpgid(pid), sys_signal.SIGTERM)
+        raise Exception
+
+
+def check_pulling(line: str, cmd_name: str, progress_bar: Progress) -> None:
+    task = progress_bar.tasks[0]
+    if "Pulling" in line and "fs layer" not in line:
+        progress_bar.update(
+            0,
+            description=f"⌛ [bold]{cmd_name} [{task.completed} / {task.total+1}]",
+            total=task.total + 1,
+            refresh=True,
+        )
+    if "Pulled" in line:
+        progress_bar.update(
+            0,
+            description=f"⌛ [bold]{cmd_name} [{task.completed + 1} / {task.total}]",
+            completed=task.completed + 1,
+            refresh=True,
+        )
+        if progress_bar.finished:
+            progress_bar.update(
+                0,
+                description=f"✅ [bold green]{cmd_name} [{task.completed} / {task.total}]",
+                refresh=True,
+            )
 
-    Args:
-        cmds (list): list of commands to be executed
-        dry_run (bool, optional): If `True` only displays cmds to be executed. Defaults to False.
-    """
-    process_list: TypeList = []
-    console = rich.get_console()
 
-    username, password = (
-        extract_username_and_pass(cmds[0]) if len(cmds) > 0 else ("-", "-")
+def check_building(line: str, cmd_name: str, progress_bar: Progress) -> None:
+
+    load_pattern = re.compile(
+        r"^#.* load build definition from [A-Za-z0-9]+\.dockerfile$", re.IGNORECASE
     )
-    # display VM credentials
-    console.print(generate_user_table(username=username, password=password))
+    build_pattern = re.compile(
+        r"^#.* naming to docker\.io/openmined/.* done$", re.IGNORECASE
+    )
+    task = progress_bar.tasks[0]
+
+    if load_pattern.match(line):
+        progress_bar.update(
+            0,
+            description=f"⌛ [bold]{cmd_name} [{task.completed} / {task.total +1}]",
+            total=task.total + 1,
+            refresh=True,
+        )
+    if build_pattern.match(line):
+        progress_bar.update(
+            0,
+            description=f"⌛ [bold]{cmd_name} [{task.completed+1} / {task.total}]",
+            completed=task.completed + 1,
+            refresh=True,
+        )
+
+    if progress_bar.finished:
+        progress_bar.update(
+            0,
+            description=f"✅ [bold green]{cmd_name} [{task.completed} / {task.total}]",
+            refresh=True,
+        )
+
+
+def check_launching(line: str, cmd_name: str, progress_bar: Progress) -> None:
+    task = progress_bar.tasks[0]
+    if "Starting" in line:
+        progress_bar.update(
+            0,
+            description=f"⌛ [bold]{cmd_name} [{task.completed} / {task.total+1}]",
+            total=task.total + 1,
+            refresh=True,
+        )
+    if "Started" in line:
+        progress_bar.update(
+            0,
+            description=f"⌛ [bold]{cmd_name} [{task.completed + 1} / {task.total}]",
+            completed=task.completed + 1,
+            refresh=True,
+        )
+        if progress_bar.finished:
+            progress_bar.update(
+                0,
+                description=f"✅ [bold green]{cmd_name} [{task.completed} / {task.total}]",
+                refresh=True,
+            )
+
 
+def read_thread_logs(
+    progress_bar: Progress, pid: int, queue: Queue, cmd_name: str
+) -> None:
+    line = queue.get()
+    line = str(line, encoding="utf-8").strip()
+    if progress_bar:
+        check_errors(line, pid, cmd_name, progress_bar=progress_bar)
+        check_pulling(line, cmd_name, progress_bar=progress_bar)
+        check_launching(line, cmd_name, progress_bar=progress_bar)
+        check_building(line, cmd_name, progress_bar=progress_bar)
+
+
+def create_thread_logs(process: subprocess.Popen) -> Queue:
+    def enqueue_output(out: Any, queue: Queue) -> None:
+        for line in iter(out.readline, b""):
+            queue.put(line)
+        out.close()
+
+    queue: Queue = Queue()
+    thread_1 = Thread(target=enqueue_output, args=(process.stdout, queue))
+    thread_2 = Thread(target=enqueue_output, args=(process.stderr, queue))
+
+    thread_1.daemon = True  # thread dies with the program
+    thread_1.start()
+    thread_2.daemon = True  # thread dies with the program
+    thread_2.start()
+    return queue
+
+
+def process_cmd(
+    cmds: TypeList[str],
+    dry_run: bool,
+    silent: bool,
+    from_rendered_dir: bool,
+    progress_bar: Union[Progress, None] = None,
+    cmd_name: str = "",
+) -> None:
+    process_list: TypeList = []
     cwd = (
         os.path.join(GRID_SRC_PATH, RENDERED_DIR)
         if from_rendered_dir
         else GRID_SRC_PATH
     )
+    username, password = (
+        extract_username_and_pass(cmds[0]) if len(cmds) > 0 else ("-", "-")
+    )
 
-    print("Current Working Directory: ", cwd)
+    # display VM credentials
+    console = rich.get_console()
+    credentials = generate_user_table(username=username, password=password)
+    if credentials:
+        console.print(credentials)
 
     for cmd in cmds:
         if dry_run:
@@ -409,15 +547,32 @@ def execute_commands(
             else:
                 display_jupyter_token(cmd)
                 if silent:
+                    ON_POSIX = "posix" in sys.builtin_module_names
+
                     process = subprocess.Popen(  # nosec
                         cmd_to_exec,
                         stdout=subprocess.PIPE,
                         stderr=subprocess.PIPE,
                         cwd=cwd,
+                        close_fds=ON_POSIX,
                         shell=True,
                     )
-                    process.communicate()
+
+                    # Creates two threads to get docker stdout and sterr
+                    logs_queue = create_thread_logs(process=process)
+
+                    while process.poll() != 0:
+                        while True:
+                            if logs_queue.empty():
+                                break
+                            # Read stdout and sterr to check errors or update progress bar.
+                            read_thread_logs(
+                                progress_bar, process.pid, logs_queue, cmd_name
+                            )
                 else:
+                    if progress_bar:
+                        progress_bar.stop()
+
                     subprocess.run(  # nosec
                         cmd_to_exec,
                         shell=True,
@@ -434,6 +589,45 @@ def execute_commands(
         save_vm_details_as_json(username, password, process_list)
 
 
+def execute_commands(
+    cmds: Union[TypeList[str], TypeDict[str, TypeList[str]]],
+    dry_run: bool = False,
+    silent: bool = False,
+    from_rendered_dir: bool = False,
+) -> None:
+    """Execute the launch commands and display their status in realtime.
+
+    Args:
+        cmds (list): list of commands to be executed
+        dry_run (bool, optional): If `True` only displays cmds to be executed. Defaults to False.
+    """
+    console = rich.get_console()
+    if isinstance(cmds, dict):
+        console.print("[bold green]⠋[bold blue] Launching Docker Images [/bold blue]\t")
+        for cmd_name, cmd in cmds.items():
+            with Progress(console=console, auto_refresh=False) as progress:
+                if silent:
+                    progress.add_task(
+                        f"[bold green]{cmd_name} Images",
+                        total=0,
+                    )
+                process_cmd(
+                    cmds=cmd,
+                    dry_run=dry_run,
+                    silent=silent,
+                    from_rendered_dir=from_rendered_dir,
+                    progress_bar=progress,
+                    cmd_name=cmd_name,
+                )
+    else:
+        process_cmd(
+            cmds=cmds,
+            dry_run=dry_run,
+            silent=silent,
+            from_rendered_dir=from_rendered_dir,
+        )
+
+
 def display_vm_status(process_list: TypeList) -> None:
     """Display the status of the processes being executed on the VM.
 
@@ -809,7 +1003,7 @@ def create_launch_cmd(
     verb: GrammarVerb,
     kwargs: TypeDict[str, Any],
     ignore_docker_version_check: Optional[bool] = False,
-) -> Union[str, TypeList[str]]:
+) -> Union[str, TypeList[str], TypeDict[str, TypeList[str]]]:
     parsed_kwargs: TypeDict[str, Any] = {}
     host_term = verb.get_named_term_hostgrammar(name="host")
     host = host_term.host
@@ -905,6 +1099,7 @@ def create_launch_cmd(
         parsed_kwargs.update(kwargs)
 
     if host in ["docker"]:
+
         # Check docker service status
         if not ignore_docker_version_check:
             check_docker_service_status()
@@ -1388,13 +1583,39 @@ def create_launch_cmd(
     )
 
 
+def pull_command(cmd: str, kwargs: TypeDict[str, Any]) -> TypeList[str]:
+    pull_cmd = str(cmd)
+    if kwargs["release"] == "production":
+        pull_cmd += " --file docker-compose.yml"
+    else:
+        pull_cmd += " --file docker-compose.pull.yml"
+    pull_cmd += " pull"
+    return [pull_cmd]
+
+
+def build_command(cmd: str) -> TypeList[str]:
+    build_cmd = str(cmd)
+    build_cmd += " --file docker-compose.build.yml"
+    build_cmd += " --file docker-compose.dev.yml"
+    build_cmd += " build"
+    return [build_cmd]
+
+
+def deploy_command(cmd: str, tail: bool) -> TypeList[str]:
+    up_cmd = str(cmd)
+    up_cmd += " up"
+    if not tail:
+        up_cmd += " -d"
+    return [up_cmd]
+
+
 def create_launch_docker_cmd(
     verb: GrammarVerb,
     docker_version: str,
     kwargs: TypeDict[str, Any],
     tail: bool = True,
     silent: bool = False,
-) -> str:
+) -> TypeDict[str, TypeList[str]]:
     host_term = verb.get_named_term_hostgrammar(name="host")
     node_name = verb.get_named_term_type(name="node_name")
     node_type = verb.get_named_term_type(name="node_type")
@@ -1415,10 +1636,10 @@ def create_launch_docker_cmd(
 
     print("  - TYPE: " + str(node_type.input))
     print("  - NAME: " + str(snake_name))
-    # print("  - TAG: " + str(tag))
+    print("  - TAG: " + str(tag))
     print("  - PORT: " + str(host_term.free_port))
     print("  - DOCKER COMPOSE: " + docker_version)
-    # print("  - TAIL: " + str(tail))
+    print("  - TAIL: " + str(tail))
     print("\n")
 
     version_string = kwargs["tag"]
@@ -1489,6 +1710,9 @@ def create_launch_docker_cmd(
     if "test" in kwargs and kwargs["test"] is True:
         envs["S3_VOLUME_SIZE_MB"] = "100"  # GitHub CI is small
 
+    if kwargs.get("release", "") == "development":
+        envs["RABBITMQ_MANAGEMENT"] = "-management"
+
     if "release" in kwargs:
         envs["RELEASE"] = kwargs["release"]
 
@@ -1507,10 +1731,6 @@ def create_launch_docker_cmd(
     else:
         cmd += " ".join(args)
 
-    if not build:
-        pull_cmd = str(cmd)
-        pull_cmd += " docker compose pull"
-
     cmd += " docker compose -p " + snake_name
 
     if "vpn" in kwargs and kwargs["vpn"]:
@@ -1568,29 +1788,21 @@ def create_launch_docker_cmd(
     except Exception:  # nosec
         pass
 
+    final_commands = {}
+    final_commands["Pulling"] = pull_command(cmd, kwargs)
+
     cmd += " --file docker-compose.yml"
-    if build:
-        cmd += " --file docker-compose.build.yml"
-    if "release" in kwargs and kwargs["release"] == "development":
-        cmd += " --file docker-compose.dev.yml"
     if "tls" in kwargs and kwargs["tls"] is True:
         cmd += " --file docker-compose.tls.yml"
     if "test" in kwargs and kwargs["test"] is True:
         cmd += " --file docker-compose.test.yml"
-    cmd += " up"
-
-    if not tail:
-        cmd += " -d"
 
     if build:
-        cmd += " --build"  # force rebuild
-    else:
-        if is_windows():
-            cmd = pull_cmd + "; " + cmd
-        else:
-            cmd = pull_cmd + " && " + cmd
+        my_build_command = build_command(cmd)
+        final_commands["Building"] = my_build_command
 
-    return cmd
+    final_commands["Launching"] = deploy_command(cmd, tail)
+    return final_commands
 
 
 def create_launch_vagrant_cmd(verb: GrammarVerb) -> str:
diff --git a/packages/hagrid/tests/hagrid/cli_test.py b/packages/hagrid/tests/hagrid/cli_test.py
index 4aa36615761..92ba20bc340 100644
--- a/packages/hagrid/tests/hagrid/cli_test.py
+++ b/packages/hagrid/tests/hagrid/cli_test.py
@@ -20,6 +20,8 @@ def test_hagrid_launch() -> None:
     verb.load_grammar(grammar=grammar)
     cmd = cli.create_launch_cmd(verb=verb, kwargs={}, ignore_docker_version_check=True)
 
+    cmd = cmd["Launching"][0]  # type: ignore
+
     # check that it's a domain by default
     assert "NODE_TYPE=domain" in cmd or "NODE_TYPE='domain'" in cmd
 
@@ -56,6 +58,7 @@ def test_hagrid_launch_without_name_with_preposition() -> None:
     verb.load_grammar(grammar=grammar)
     cmd = cli.create_launch_cmd(verb=verb, kwargs={}, ignore_docker_version_check=True)
 
+    cmd = cmd["Launching"][0]  # type: ignore
     # check that it's a domain by default
     assert "NODE_TYPE=domain" in cmd or "NODE_TYPE='domain'" in cmd
 
@@ -92,7 +95,7 @@ def test_launch_with_multiword_domain_name() -> None:
     verb.load_grammar(grammar=grammar)
     cmd = cli.create_launch_cmd(verb=verb, kwargs={}, ignore_docker_version_check=True)
 
-    print(cmd)
+    cmd = cmd["Launching"][0]  # type: ignore
 
     # check that it's a domain by default
     assert "NODE_TYPE=domain" in cmd or "NODE_TYPE='domain'" in cmd
@@ -116,7 +119,7 @@ def test_launch_with_longer_multiword_domain_name() -> None:
     verb.load_grammar(grammar=grammar)
     cmd = cli.create_launch_cmd(verb=verb, kwargs={}, ignore_docker_version_check=True)
 
-    print(cmd)
+    cmd = cmd["Launching"][0]  # type: ignore
 
     # check that it's a domain by default
     assert "NODE_TYPE=domain" in cmd or "NODE_TYPE='domain'" in cmd
@@ -143,7 +146,7 @@ def test_launch_with_longer_multiword_domain_name_with_preposition() -> None:
     verb.load_grammar(grammar=grammar)
     cmd = cli.create_launch_cmd(verb=verb, kwargs={}, ignore_docker_version_check=True)
 
-    print(cmd)
+    cmd = cmd["Launching"][0]  # type: ignore
 
     # check that it's a domain by default
     assert "NODE_TYPE=domain" in cmd or "NODE_TYPE='domain'" in cmd


Code review comments:
diff_hunk: @@ -409,15 +547,32 @@ def execute_commands(
             else:
                 display_jupyter_token(cmd)
                 if silent:
+                    ON_POSIX = "posix" in sys.builtin_module_names
+
                     process = subprocess.Popen(  # nosec
                         cmd_to_exec,
                         stdout=subprocess.PIPE,
                         stderr=subprocess.PIPE,
                         cwd=cwd,
+                        close_fds=ON_POSIX,
                         shell=True,
                     )
-                    process.communicate()
+
+                    # Creates two threads to get docker stdout and sterr
+                    logs_queue = create_thread_logs(process=process)
+
+                    while process.poll() != 0:
+                        while True:
path: packages/hagrid/hagrid/cli.py
text: @user1:
NIT: We could replace True by `not logs_queue.empty()` ?

@author:
Solved by: [c68b734](https://github.com/OpenMined/PySyft/pull/6900/commits/c68b734ac04b47b6d265f903532febdc2aede8e7)