Merge pull request #640 from golemfactory/scx1332/stability

Scx1332/stability

goth/assertions/monitor.py

@@ -379,7 +379,7 @@ async def wait_for_event(
         Subsequent calls will examine all events gathered since the previous call
         returned and then wait for up to `timeout` seconds.
 
-        When `timeout` elapses, `asyncio.TimeourError` will be raised.
+        When `timeout` elapses, `asyncio.TimeoutError` will be raised.
         """
 
         # First examine log lines already seen

goth/runner/probe/__init__.py

@@ -11,6 +11,7 @@
 import shlex
 import signal
 import traceback
+from time import perf_counter
 from typing import (
     AsyncIterator,
     Dict,
@@ -22,6 +23,7 @@
     TYPE_CHECKING,
 )
 
+import aiohttp
 from docker import DockerClient
 
 from goth.address import (
@@ -224,6 +226,43 @@ def remove(self) -> None:
             self.container.remove(force=True)
             self._logger.debug("Container removed")
 
+    async def _wait_for_yagna_start(self, timeout: float = 30) -> None:
+        host_yagna_addr = f"http://127.0.0.1:{self.container.ports[YAGNA_REST_PORT]}"
+        self._logger.info(f"Waiting for yagna REST API: {host_yagna_addr}")
+        self._logger.info(
+            f"Waiting for yagna http endpoint: {host_yagna_addr}, timeout: {timeout:.1f}"
+        )
+        start_time = perf_counter()
+        async with aiohttp.ClientSession() as session:
+            while True:
+                try:
+                    async with session.get(f"{host_yagna_addr}/version/get") as resp:
+                        yagna_status_obj = await resp.json()
+                        yagna_version = yagna_status_obj["current"]["version"]
+                        elapsed = perf_counter() - start_time
+                        self._logger.info(
+                            f"Yagna responded with version: {yagna_version}"
+                            f" after {elapsed:.1f}/{timeout:.1f} seconds"
+                        )
+                        if timeout - elapsed < 5:
+                            self._logger.warning(
+                                f"Only {timeout - elapsed:.1f} seconds left to timeout. "
+                                "Consider using a higher timeout."
+                            )
+                        return yagna_version
+                except aiohttp.ClientConnectionError as ex:
+                    self._logger.debug(f"Failed to connect to yagna - trying again: {ex}")
+                    pass
+
+                elapsed = perf_counter() - start_time
+                if elapsed > timeout:
+                    raise Exception(
+                        f"Timeout {timeout} exceeded: "
+                        f"Failed to get data from endpoint: {host_yagna_addr}"
+                    )
+
+                await asyncio.sleep(0.5)
+
     async def _start_container(self) -> None:
         """
         Start the probe's Docker container.
@@ -233,22 +272,14 @@ async def _start_container(self) -> None:
         """
         self.container.start()
 
-        # Wait until the daemon is ready to create an app key.
-        self._logger.info("Waiting for connection to ya-sb-router")
-        if self.container.logs:
-            await self.container.logs.wait_for_entry(
-                ".*connected with server: ya-sb-router.*", timeout=30
-            )
+        await self._wait_for_yagna_start(60)
+
         await self.create_app_key()
 
-        self._logger.info("Waiting for yagna REST API to be listening")
-        if self.container.logs:
-            await self.container.logs.wait_for_entry(
-                "Starting .*actix-web-service.* service on .*."
-                "|"
-                "Http server thread started on:.*",
-                timeout=30,
-            )
+        # restart container to allow faster discovery of new identity in the network
+        self._logger.info("Restarting container after identity set")
+        self.container.restart()
+        await self._wait_for_yagna_start(60)
 
         # Obtain the IP address of the container
         self.ip_address = get_container_address(self._docker_client, self.container.name)
@@ -293,8 +324,6 @@ async def create_app_key(self, key_name: str = "test_key") -> str:
                 raise
             db_id = self.cli.id_update(address, set_default=True)
             self._logger.debug("update_id. result=%r", db_id)
-            self.container.restart()
-            await asyncio.sleep(5)
         try:
             key = self.cli.app_key_create(key_name)
             self._logger.debug("create_app_key. key_name=%s, key=%s", key_name, key)