feat: add healthcheck

.gitleaksignore

@@ -21,4 +21,9 @@ e7de9ce0b902ed6d68f8c5b033d044f39b08f5a1:operate/data/contracts/service_staking_
 d8149e9b5b7bd6a7ed7bc1039900702f1d4f287b:operate/services/manage.py:generic-api-key:405
 d8149e9b5b7bd6a7ed7bc1039900702f1d4f287b:operate/services/manage.py:generic-api-key:406
 d8149e9b5b7bd6a7ed7bc1039900702f1d4f287b:operate/services/manage.py:generic-api-key:454
-d8149e9b5b7bd6a7ed7bc1039900702f1d4f287b:operate/services/manage.py:generic-api-key:455
+d8149e9b5b7bd6a7ed7bc1039900702f1d4f287b:operate/services/manage.py:generic-api-key:45591ec07457f69e9a29f63693ac8ef887e4b5f49f0:operate/services/manage.py:generic-api-key:454
+99c0f139b037da2587708212fcf6d0e20786d0ba:operate/services/manage.py:generic-api-key:405
+99c0f139b037da2587708212fcf6d0e20786d0ba:operate/services/manage.py:generic-api-key:406
+99c0f139b037da2587708212fcf6d0e20786d0ba:operate/services/manage.py:generic-api-key:454
+99c0f139b037da2587708212fcf6d0e20786d0ba:operate/services/manage.py:generic-api-key:455
+91ec07457f69e9a29f63693ac8ef887e4b5f49f0:operate/services/manage.py:generic-api-key:454

operate/cli.py

@@ -145,6 +145,7 @@ def create_app(  # pylint: disable=too-many-locals, unused-argument, too-many-st
     logger = setup_logger(name="operate")
     operate = OperateApp(home=home, logger=logger)
     funding_jobs: t.Dict[str, asyncio.Task] = {}
+    healthcheck_jobs: t.Dict[str, asyncio.Task] = {}
 
     # Create shutdown endpoint
     shutdown_endpoint = uuid.uuid4().hex
@@ -171,6 +172,22 @@ def schedule_funding_job(
             )
         )
 
+    def schedule_healthcheck_job(
+        service: str,
+    ) -> None:
+        """Schedule a healthcheck job."""
+        logger.info(f"Starting healthcheck job for {service}")
+        if service in healthcheck_jobs:
+            logger.info(f"Cancelling existing healthcheck_jobs job for {service}")
+            cancel_healthcheck_job(service=service)
+
+        loop = asyncio.get_running_loop()
+        healthcheck_jobs[service] = loop.create_task(
+            operate.service_manager().healthcheck_job(
+                hash=service,
+            )
+        )
+
     def cancel_funding_job(service: str) -> None:
         """Cancel funding job."""
         if service not in funding_jobs:
@@ -179,6 +196,14 @@ def cancel_funding_job(service: str) -> None:
         if not status:
             logger.info(f"Funding job cancellation for {service} failed")
 
+    def cancel_healthcheck_job(service: str) -> None:
+        """Cancel healthcheck job."""
+        if service not in healthcheck_jobs:
+            return
+        status = healthcheck_jobs[service].cancel()
+        if not status:
+            logger.info(f"Healthcheck job cancellation for {service} failed")
+
     app = FastAPI()
 
     app.add_middleware(
@@ -506,6 +531,7 @@ async def _create_services(request: Request) -> JSONResponse:
             manager.fund_service(hash=service.hash)
             manager.deploy_service_locally(hash=service.hash)
             schedule_funding_job(service=service.hash)
+            schedule_healthcheck_job(service=service.hash)
 
         return JSONResponse(
             content=operate.service_manager().create_or_load(hash=service.hash).json
@@ -529,6 +555,7 @@ async def _update_services(request: Request) -> JSONResponse:
             manager.fund_service(hash=service.hash)
             manager.deploy_service_locally(hash=service.hash)
             schedule_funding_job(service=service.hash)
+            schedule_healthcheck_job(service=service.hash)
 
         return JSONResponse(content=service.json)
 
@@ -638,6 +665,7 @@ async def _start_service_locally(request: Request) -> JSONResponse:
         manager.fund_service(hash=service)
         manager.deploy_service_locally(hash=service, force=True)
         schedule_funding_job(service=service)
+        schedule_healthcheck_job(service=service.hash)
         return JSONResponse(content=manager.create_or_load(service).deployment)
 
     @app.post("/api/services/{service}/deployment/stop")

operate/services/manage.py

@@ -26,6 +26,7 @@
 from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
 
+import aiohttp  # type: ignore
 from aea.helpers.base import IPFSHash
 from aea.helpers.logging import setup_logger
 from autonomy.chain.base import registry_contracts
@@ -56,6 +57,18 @@
 KEYS_JSON = "keys.json"
 DOCKER_COMPOSE_YAML = "docker-compose.yaml"
 SERVICE_YAML = "service.yaml"
+HTTP_OK = 200
+
+
+async def check_service_health() -> bool:
+    """Check the service health"""
+    async with aiohttp.ClientSession() as session:
+        async with session.get("http://localhost:8000/healthcheck") as resp:
+            status = resp.status
+            response_json = await resp.json()
+            return status == HTTP_OK and response_json.get(
+                "is_transitioning_fast", False
+            )
 
 
 class ServiceManager:
@@ -853,6 +866,32 @@ async def funding_job(
                     )
                 await asyncio.sleep(60)
 
+    async def healthcheck_job(
+        self,
+        hash: str,
+    ) -> None:
+        """Start a background funding job."""
+        failed_health_checks = 0
+
+        while True:
+            try:
+                # Check the service health
+                healthy = await check_service_health()
+                # Restart the service if the health failed 5 times in a row
+                if not healthy:
+                    failed_health_checks += 1
+                else:
+                    failed_health_checks = 0
+                if failed_health_checks >= 5:
+                    self.stop_service_locally(hash=hash)
+                    self.deploy_service_locally(hash=hash)
+
+            except Exception:  # pylint: disable=broad-except
+                logging.info(
+                    f"Error occured while checking the service health\n{traceback.format_exc()}"
+                )
+            await asyncio.sleep(60)
+
     def deploy_service_locally(self, hash: str, force: bool = True) -> Deployment:
         """
         Deploy service locally

operate/services/protocol.py

@@ -680,7 +680,7 @@ def swap(  # pylint: disable=too-many-arguments,too-many-locals
             key_file = Path(temp_dir, "key.txt")
             key_file.write_text(owner_key, encoding="utf-8")
             owner_crypto = EthereumCrypto(private_key_path=str(key_file))
-        owner_cryptos: list[EthereumCrypto] = [owner_crypto]
+        owner_cryptos: t.List[EthereumCrypto] = [owner_crypto]
         owners = [
             manager.ledger_api.api.to_checksum_address(owner_crypto.address)
             for owner_crypto in owner_cryptos