services API + simple toy service example

examples/erigon/erigon.py

@@ -0,0 +1,108 @@
+import asyncio
+
+from dataclasses import dataclass
+
+from yapapi.props.base import prop, constraint
+from yapapi.props import inf
+
+from yapapi.payload import Payload
+from yapapi.executor import Golem
+from yapapi.executor.services import Service
+
+from yapapi.log import enable_default_logger, log_summary, log_event_repr  # noqa
+
+
+TURBOGETH_RUNTIME_NAME = "turbogeth-managed"
+PROP_ERIGON_ETHEREUM_NETWORK = "golem.srv.app.eth.network"
+
+
+@dataclass
+class ErigonPayload(Payload):
+    network: str = prop(PROP_ERIGON_ETHEREUM_NETWORK)
+
+    runtime: str = constraint(inf.INF_RUNTIME_NAME, default=TURBOGETH_RUNTIME_NAME)
+    min_mem_gib: float = constraint(inf.INF_MEM, operator=">=", default=16)
+    min_storage_gib: float = constraint(inf.INF_STORAGE, operator=">=", default=1024)
+
+
+class ErigonService(Service):
+    credentials = None
+
+    def post_init(self):
+        self.credentials = {}
+
+    def __repr__(self):
+        srv_repr = super().__repr__()
+        return f"{srv_repr}, credentials: {self.credentials}"
+
+    @staticmethod
+    async def get_payload():
+        return ErigonPayload(network="rinkeby")
+
+    async def start(self):
+        deploy_idx = self.ctx.deploy()
+        self.ctx.start()
+        future_results = yield self.ctx.commit()
+        results = await future_results
+        self.credentials = "RECEIVED" or results[deploy_idx]  # (NORMALLY, WOULD BE PARSED)
+
+    async def run(self):
+
+        while True:
+            print(f"service {self.ctx.id} running on {self.ctx.provider_name} ... ")
+            signal = self._listen_nowait()
+            if signal and signal.message == "go":
+                self.ctx.run("go!")
+                yield self.ctx.commit()
+            else:
+                await asyncio.sleep(1)
+                yield
+
+    async def shutdown(self):
+        self.ctx.download_file("some/service/state", "temp/path")
+        yield self.ctx.commit()
+
+
+async def main(subnet_tag, driver=None, network=None):
+
+    async with Golem(
+        budget=10.0,
+        subnet_tag=subnet_tag,
+        driver=driver,
+        network=network,
+        event_consumer=log_summary(log_event_repr),
+    ) as golem:
+        cluster = await golem.run_service(
+            ErigonService,
+            num_instances=1,
+        )
+
+        def instances():
+            return [{s.ctx.id, s.state.value} for s in cluster.instances]
+
+        def still_running():
+            return any([s for s in cluster.instances if s.is_available])
+
+        cnt = 0
+        while cnt < 10:
+            print(f"instances: {instances()}")
+            await asyncio.sleep(3)
+            cnt += 1
+            if cnt == 3:
+                if len(cluster.instances) > 1:
+                    cluster.instances[0].send_message_nowait("go")
+
+        for s in cluster.instances:
+            cluster.stop_instance(s)
+
+        print(f"instances: {instances()}")
+
+        cnt = 0
+        while cnt < 10 and still_running():
+            print(f"instances: {instances()}")
+            await asyncio.sleep(1)
+
+    print(f"instances: {instances()}")
+
+
+asyncio.run(main(None))

pyproject.toml

@@ -46,6 +46,7 @@ colorama = "^0.4.4"
 # would not work: see https://github.com/python-poetry/poetry/issues/129.
 goth = { version = "^0.2.1", optional = true, python = "^3.8.0" }
 Deprecated = "^1.2.12"
+python-statemachine = "^0.8.0"
 
 [tool.poetry.extras]
 cli = ['fire', 'rich']

yapapi/executor/__init__.py

@@ -22,7 +22,9 @@
     Optional,
     Set,
     Tuple,
+    Type,
     TypeVar,
+    TYPE_CHECKING,
     Union,
     cast,
     overload,
@@ -48,6 +50,9 @@
 from ..rest.market import OfferProposal, Subscription
 from ..storage import gftp
 from ._smartq import Consumer, Handle, SmartQueue
+
+if TYPE_CHECKING:
+    from .services import Cluster, Service
 from .strategy import (
     DecreaseScoreForUnconfirmedAgreement,
     LeastExpensiveLinearPayuMS,
@@ -566,6 +571,33 @@ async def execute_tasks(
             async for t in executor.submit(worker, data):
                 yield t
 
+    async def run_service(
+        self,
+        service_class: Type["Service"],
+        num_instances: int = 1,
+        payload: Optional[Payload] = None,
+        expiration: Optional[datetime] = None,
+    ) -> "Cluster":
+        from .services import Cluster  # avoid circular dependency
+
+        payload = payload or await service_class.get_payload()
+
+        if not payload:
+            raise ValueError(
+                f"No payload returned from {service_class.__name__}.get_payload() nor given in the `payload` argument."
+            )
+
+        cluster = Cluster(
+            engine=self,
+            service_class=service_class,
+            payload=payload,
+            num_instances=num_instances,
+            expiration=expiration,
+        )
+        await self._stack.enter_async_context(cluster)
+        cluster.spawn_instances()
+        return cluster
+
 
 class Job:
     """Functionality related to a single job."""

yapapi/executor/ctx.py

@@ -51,9 +51,13 @@ def timeout(self) -> Optional[timedelta]:
         return None
 
 
-class _InitStep(Work):
+class _Deploy(Work):
     def register(self, commands: CommandContainer):
         commands.deploy()
+
+
+class _Start(Work):
+    def register(self, commands: CommandContainer):
         commands.start()
 
 
@@ -277,27 +281,36 @@ def __init__(
         node_info: NodeInfo,
         storage: StorageProvider,
         emitter: Optional[Callable[[StorageEvent], None]] = None,
+        implicit_init: bool = True,
     ):
         self.id = ctx_id
         self._node_info = node_info
         self._storage: StorageProvider = storage
         self._pending_steps: List[Work] = []
         self._started: bool = False
         self._emitter: Optional[Callable[[StorageEvent], None]] = emitter
+        self._implicit_init = implicit_init
 
     @property
     def provider_name(self) -> Optional[str]:
         """Return the name of the provider associated with this work context."""
         return self._node_info.name
 
     def __prepare(self):
-        if not self._started:
-            self._pending_steps.append(_InitStep())
+        if not self._started and self._implicit_init:
+            self.deploy()
+            self.start()
             self._started = True
 
     def begin(self):
         pass
 
+    def deploy(self):
+        self._pending_steps.append(_Deploy())
+
+    def start(self):
+        self._pending_steps.append(_Start())
+
     def send_json(self, json_path: str, data: dict):
         """Schedule sending JSON data to the provider.