simple service toy prototype

examples/simple-service-poc/simple_service.py

@@ -0,0 +1,181 @@
+#!/usr/bin/env python3
+"""
+the requestor agent controlling and interacting with the "simple service"
+"""
+import asyncio
+from datetime import datetime, timedelta, timezone
+import json
+import pathlib
+import queue
+import random
+import string
+import sys
+
+
+
+from yapapi import (
+    NoPaymentAccountError,
+    __version__ as yapapi_version,
+    windows_event_loop_fix,
+)
+from yapapi.executor import Golem
+from yapapi.executor.services import Service
+
+from yapapi.log import enable_default_logger, log_summary, log_event_repr, pluralize  # noqa
+from yapapi.payload import vm
+
+examples_dir = pathlib.Path(__file__).resolve().parent.parent
+sys.path.append(str(examples_dir))
+
+from utils import (
+    build_parser,
+    TEXT_COLOR_CYAN,
+    TEXT_COLOR_DEFAULT,
+    TEXT_COLOR_RED,
+    TEXT_COLOR_YELLOW,
+)
+
+NUM_INSTANCES = 1
+
+
+class SimpleService(Service):
+    STATS_PATH = "/golem/out/stats"
+    PLOT_INFO_PATH = "/golem/out/plot"
+    SIMPLE_SERVICE = "/golem/run/simple_service.py"
+
+    @staticmethod
+    async def get_payload():
+        return await vm.repo(
+        image_hash="8b11df59f84358d47fc6776d0bb7290b0054c15ded2d6f54cf634488",
+        min_mem_gib=0.5,
+        min_storage_gib=2.0,
+    )
+
+    async def start(self):
+        self._ctx.run("/golem/run/simulate_observations_ctl.py", "--start")
+        yield self._ctx.commit()
+
+    async def run(self):
+        while True:
+            await asyncio.sleep(10)
+            self._ctx.run(self.SIMPLE_SERVICE, "--stats")  # idx 0
+            self._ctx.run(self.SIMPLE_SERVICE, "--plot", "dist")  # idx 1
+
+            future_results = yield self._ctx.commit()
+            results = await future_results
+            stats = results[0].stdout.strip()
+            plot = results[1].stdout.strip().strip('"')
+
+            print(f"{TEXT_COLOR_CYAN}stats: {stats}{TEXT_COLOR_DEFAULT}")
+
+            plot_filename = (
+                    "".join(random.choice(string.ascii_letters) for _ in
+                            range(10)) + ".png"
+            )
+            print(f"{TEXT_COLOR_CYAN}downloading plot: {plot} to {plot_filename}{TEXT_COLOR_DEFAULT}")
+            self._ctx.download_file(plot, str(pathlib.Path(__file__).resolve().parent / plot_filename))
+
+            steps = self._ctx.commit()
+            yield steps
+
+    async def shutdown(self):
+        self._ctx.run("/golem/run/simulate_observations_ctl.py", "--stop")
+        yield self._ctx.commit()
+
+
+async def main(subnet_tag, driver=None, network=None):
+    async with Golem(
+        budget=1.0,
+        subnet_tag=subnet_tag,
+        driver=driver,
+        network=network,
+        event_consumer=log_summary(log_event_repr),
+    ) as golem:
+
+        print(
+            f"yapapi version: {TEXT_COLOR_YELLOW}{yapapi_version}{TEXT_COLOR_DEFAULT}\n"
+            f"Using subnet: {TEXT_COLOR_YELLOW}{subnet_tag}{TEXT_COLOR_DEFAULT}, "
+            f"payment driver: {TEXT_COLOR_YELLOW}{golem.driver}{TEXT_COLOR_DEFAULT}, "
+            f"and network: {TEXT_COLOR_YELLOW}{golem.network}{TEXT_COLOR_DEFAULT}\n"
+        )
+
+        start_time = datetime.now()
+
+        print(f"{TEXT_COLOR_YELLOW}starting {pluralize(NUM_INSTANCES, 'instance')}{TEXT_COLOR_DEFAULT}")
+
+        cluster = await golem.run_service(
+            SimpleService,
+            num_instances=NUM_INSTANCES,
+            expiration=datetime.now(timezone.utc) + timedelta(minutes=15))
+
+        def instances():
+            return [(s.provider_name, s.state.value) for s in cluster.instances]
+
+        def still_running():
+            return any([s for s in cluster.instances if s.is_available])
+
+        while datetime.now() < start_time + timedelta(minutes=2):
+            print(f"instances: {instances()}")
+            await asyncio.sleep(5)
+
+        print(f"{TEXT_COLOR_YELLOW}stopping instances{TEXT_COLOR_DEFAULT}")
+        cluster.stop()
+
+        cnt = 0
+        while cnt < 10 and still_running():
+            print(f"instances: {instances()}")
+            await asyncio.sleep(5)
+
+    print(f"instances: {instances()}")
+
+
+if __name__ == "__main__":
+    parser = build_parser("Test http")
+    now = datetime.now().strftime("%Y-%m-%d_%H.%M.%S")
+    parser.set_defaults(log_file=f"simple-service-yapapi-{now}.log")
+    args = parser.parse_args()
+
+    # This is only required when running on Windows with Python prior to 3.8:
+    windows_event_loop_fix()
+
+    enable_default_logger(
+        log_file=args.log_file,
+        debug_activity_api=True,
+        debug_market_api=True,
+        debug_payment_api=True,
+    )
+
+    loop = asyncio.get_event_loop()
+    task = loop.create_task(
+        main(subnet_tag=args.subnet_tag, driver=args.driver, network=args.network)
+    )
+
+    try:
+        loop.run_until_complete(task)
+    except NoPaymentAccountError as e:
+        handbook_url = (
+            "https://handbook.golem.network/requestor-tutorials/"
+            "flash-tutorial-of-requestor-development"
+        )
+        print(
+            f"{TEXT_COLOR_RED}"
+            f"No payment account initialized for driver `{e.required_driver}` "
+            f"and network `{e.required_network}`.\n\n"
+            f"See {handbook_url} on how to initialize payment accounts for a requestor node."
+            f"{TEXT_COLOR_DEFAULT}"
+        )
+    except KeyboardInterrupt:
+        print(
+            f"{TEXT_COLOR_YELLOW}"
+            "Shutting down gracefully, please wait a short while "
+            "or press Ctrl+C to exit immediately..."
+            f"{TEXT_COLOR_DEFAULT}"
+        )
+        task.cancel()
+        try:
+            loop.run_until_complete(task)
+            print(
+                f"{TEXT_COLOR_YELLOW}Shutdown completed, thank you for waiting!{TEXT_COLOR_DEFAULT}"
+            )
+        except (asyncio.CancelledError, KeyboardInterrupt):
+            pass

examples/simple-service-poc/simple_service/simple_service.Dockerfile

@@ -0,0 +1,9 @@
+FROM python:3.8-slim
+VOLUME /golem/in /golem/out
+COPY simple_service.py /golem/run/simple_service.py
+COPY simulate_observations.py /golem/run/simulate_observations.py
+COPY simulate_observations_ctl.py /golem/run/simulate_observations_ctl.py
+RUN pip install numpy matplotlib
+RUN chmod +x /golem/run/*
+RUN /golem/run/simple_service.py --init
+ENTRYPOINT ["sh"]

examples/simple-service-poc/simple_service/simple_service.py

@@ -0,0 +1,129 @@
+#!/usr/local/bin/python
+"""
+a very basic "stub" that exposes a few commands of an imagined, very simple CLI-based
+service that is able to accumulate some linear, time-based values and present it stats
+(characteristics of the statistical distribution of the data collected so far) or provide
+distribution and time-series plots of the collected data
+"""
+import argparse
+from datetime import datetime
+import enum
+import contextlib
+import json
+import matplotlib.pyplot as plt
+import numpy
+import random
+import sqlite3
+import string
+from pathlib import Path
+
+DB_PATH = Path(__file__).absolute().parent / "service.db"
+PLOT_PATH = Path("/golem/out").absolute()
+
+
+class PlotType(enum.Enum):
+    time = "time"
+    dist = "dist"
+
+
+@contextlib.contextmanager
+def _connect_db():
+    db = sqlite3.connect(DB_PATH)
+    db.row_factory = sqlite3.Row
+    try:
+        yield db.cursor()
+        db.commit()
+    finally:
+        db.close()
+
+
+def init():
+    with _connect_db() as db:
+        db.execute(
+            "create table observations("
+            "id integer primary key autoincrement not null, "
+            "val float not null,"
+            "time_added timestamp default current_timestamp not null"
+            ")"
+        )
+
+
+def add(val):
+    with _connect_db() as db:
+        db.execute("insert into observations (val) values (?)", [val])
+
+
+def plot(plot_type):
+    data = _get_data()
+
+    if not data:
+        print(json.dumps(""))
+        return
+
+    y = [r["val"] for r in data]
+
+    if plot_type == PlotType.dist.value:
+        plt.hist(y)
+    elif plot_type == PlotType.time.value:
+        x = [datetime.strptime(r["time_added"], "%Y-%m-%d %H:%M:%S") for r in data]
+        plt.plot(x, y)
+
+    plot_filename = PLOT_PATH / (
+        "".join(random.choice(string.ascii_letters) for _ in range(10)) + ".png"
+    )
+    plt.savefig(plot_filename)
+    print(json.dumps(str(plot_filename)))
+
+
+def dump():
+    print(json.dumps(_get_data()))
+
+
+def _get_data():
+    with _connect_db() as db:
+        db.execute("select val, time_added from observations order by time_added asc")
+        return list(map(dict, db.fetchall()))
+
+
+def _get_stats(data=None):
+    data = data or [r["val"] for r in _get_data()]
+    return {
+        "min": min(data) if data else None,
+        "max": max(data) if data else None,
+        "median": numpy.median(data) if data else None,
+        "mean": numpy.mean(data) if data else None,
+        "variance": numpy.var(data) if data else None,
+        "std dev": numpy.std(data) if data else None,
+        "size": len(data),
+    }
+
+
+def stats():
+    print(json.dumps(_get_stats()))
+
+
+def get_arg_parser():
+    parser = argparse.ArgumentParser(description="simple service")
+    group = parser.add_mutually_exclusive_group(required=True)
+    group.add_argument("--add", type=float)
+    group.add_argument("--init", action="store_true")
+    group.add_argument("--plot", choices=[pt.value for pt in list(PlotType)])
+    group.add_argument("--dump", action="store_true")
+    group.add_argument("--stats", action="store_true")
+    return parser
+
+
+if __name__ == "__main__":
+    arg_parser = get_arg_parser()
+    args = arg_parser.parse_args()
+
+    if args.init:
+        init()
+    elif args.add:
+        add(args.add)
+    elif args.plot:
+        plot(args.plot)
+    elif args.dump:
+        dump()
+    elif args.stats:
+        stats()

examples/simple-service-poc/simple_service/simulate_observations.py

@@ -0,0 +1,24 @@
+#!/usr/local/bin/python
+"""
+the "hello world" service here just adds randomized numbers with normal distribution
+
+in a real-world example, this could be e.g. a thermometer connected to the provider's
+machine providing its inputs into the database or some other piece of information
+from some external source that changes over time and which can be expressed as a
+singular value
+"""
+import os
+from pathlib import Path
+import random
+import time
+
+MU = 14
+SIGMA = 3
+
+SERVICE_PATH = Path(__file__).absolute().parent / "simple_service.py"
+
+
+while True:
+    v = random.normalvariate(MU, SIGMA)
+    os.system(f"{SERVICE_PATH} --add {v}")
+    time.sleep(1)

examples/simple-service-poc/simple_service/simulate_observations_ctl.py

@@ -0,0 +1,33 @@
+#!/usr/local/bin/python
+"""
+a helper, control script that starts and stops our example `simulate_observations` service
+"""
+import argparse
+import os
+import subprocess
+import signal
+
+PIDFILE = "/var/run/simulate_observations.pid"
+SCRIPT_FILE = "/golem/run/simulate_observations.py"
+
+parser = argparse.ArgumentParser("start/stop simulation")
+group = parser.add_mutually_exclusive_group(required=True)
+group.add_argument("--start", action="store_true")
+group.add_argument("--stop", action="store_true")
+
+args = parser.parse_args()
+
+if args.start:
+    if os.path.exists(PIDFILE):
+        raise Exception(f"Cannot start process, {PIDFILE} exists.")
+    p = subprocess.Popen([SCRIPT_FILE])
+    with open(PIDFILE, "w") as pidfile:
+        pidfile.write(str(p.pid))
+elif args.stop:
+    if not os.path.exists(PIDFILE):
+        raise Exception(f"Could not find pidfile: {PIDFILE}.")
+    with open(PIDFILE, "r") as pidfile:
+        pid = int(pidfile.read())
+
+    os.kill(pid, signal.SIGKILL)
+    os.remove(PIDFILE)

yapapi/executor/ctx.py

@@ -306,9 +306,11 @@ def begin(self):
         pass
 
     def deploy(self):
+        self._implicit_init = False
         self._pending_steps.append(_Deploy())
 
     def start(self):
+        self._implicit_init = False
         self._pending_steps.append(_Start())
 
     def send_json(self, json_path: str, data: dict):