Implement persistent, local reputation + integration with the external Reputation System

pyproject.toml

@@ -24,9 +24,13 @@ appdirs = "^1.4.4"
 websocat = "^1.12.0"
 golem-node = {version = "0.15.0a18", source = "test-pypi"}
 
-setuptools = "*"
+tortoise-orm = "^0.20.0"
+aerich = "^0.7.2"
+prettytable = "^3.10.0"
 psutil = "^5.9.8"
 
+setuptools = "*"
+
 [tool.poetry.group.ray.dependencies]
 python = "^3.8.1"
 
@@ -44,6 +48,7 @@ appdirs = "^1.4.4"
 
 [tool.poetry.scripts]
 ray-on-golem = "ray_on_golem.main:main"
+rog-db = "ray_on_golem.reputation.admin:admin"
 
 [tool.poetry.group.dev.dependencies]
 poethepoet = "^0.22.0"
@@ -56,6 +61,8 @@ gvmkit-build = "^0.3.13"
 dpath = "^2.1.6"
 yamlpath = "^3.8.1"
 pytest = "^7.4.3"
+pytest-asyncio = "^0.23.5.post1"
+factory-boy = "^3.3.0"
 
 [[tool.poetry.source]]
 name = "test-pypi"

ray_on_golem/cli.py

@@ -0,0 +1,13 @@
+from pathlib import Path
+
+import click
+
+
+def with_datadir(cli_func):
+    from ray_on_golem.server.settings import DEFAULT_DATADIR
+
+    return click.option(
+        "--datadir",
+        type=Path,
+        help=f"Ray on Golem's data directory. By default, uses a system data directory: {DEFAULT_DATADIR}",
+    )(cli_func)

ray_on_golem/main.py

@@ -1,6 +1,7 @@
 import click
 
 from ray_on_golem.network_stats import main as network_stats
+from ray_on_golem.reputation.cli import reputation_cli
 from ray_on_golem.server import main as webserver
 from ray_on_golem.server import start, status, stop
 from ray_on_golem.version import get_version, version
@@ -18,6 +19,7 @@ def cli():
 cli.add_command(start)
 cli.add_command(stop)
 cli.add_command(status)
+cli.add_command(reputation_cli)
 
 
 def main():

ray_on_golem/network_stats/main.py

@@ -8,10 +8,12 @@
 import click
 import yaml
 
+from ray_on_golem.cli import with_datadir
 from ray_on_golem.network_stats.services import NetworkStatsService
 from ray_on_golem.provider.node_provider import GolemNodeProvider
+from ray_on_golem.reputation.service import ReputationService
 from ray_on_golem.server.services import YagnaService
-from ray_on_golem.server.settings import DEFAULT_DATADIR, YAGNA_PATH, get_logging_config
+from ray_on_golem.server.settings import YAGNA_PATH, get_logging_config
 
 
 def validate_config_file(ctx, param, value):
@@ -54,12 +56,7 @@ def validate_node_type(ctx, param, value):
     default=False,
     help="Enable verbose logging.",
 )
-@click.option(
-    "--datadir",
-    type=pathlib.Path,
-    help=f"Ray on Golem's data directory. [default: {DEFAULT_DATADIR}"
-    " (unless `webserver_datadir` is defined in the cluster config file)]",
-)
+@with_datadir
 def main(*args, **kwargs):
     asyncio.run(_network_stats(*args, **kwargs))
 
@@ -100,15 +97,16 @@ async def network_stats_service(
         datadir=datadir,
     )
 
-    await yagna_service.init()
-    await yagna_service.run_payment_fund(network, driver)
+    async with ReputationService(datadir):
+        await yagna_service.init()
+        await yagna_service.run_payment_fund(network, driver)
 
-    await service.init(yagna_appkey=yagna_service.yagna_appkey)
+        await service.init(yagna_appkey=yagna_service.yagna_appkey)
 
-    yield service
+        yield service
 
-    await service.shutdown()
-    await yagna_service.shutdown()
+        await service.shutdown()
+        await yagna_service.shutdown()
 
 
 if __name__ == "__main__":

ray_on_golem/network_stats/services/network_stats.py

@@ -6,7 +6,6 @@
 from typing import Dict, Optional, Sequence
 
 from golem.managers import (
-    BlacklistProviderIdPlugin,
     DefaultPaymentManager,
     DefaultProposalManager,
     NegotiatingPlugin,
@@ -26,12 +25,12 @@
 from golem.resources.proposal.exceptions import ProposalRejected
 from ya_market import ApiException
 
+from ray_on_golem.reputation.plugins import ProviderBlacklistPlugin
 from ray_on_golem.server.models import NodeConfigData
 from ray_on_golem.server.services.golem.golem import DEFAULT_DEMAND_LIFETIME
 from ray_on_golem.server.services.golem.helpers.demand_config import DemandConfigHelper
 from ray_on_golem.server.services.golem.helpers.manager_stack import ManagerStackNodeConfigHelper
 from ray_on_golem.server.services.golem.manager_stack import ManagerStack
-from ray_on_golem.server.services.golem.provider_data import PROVIDERS_BLACKLIST
 
 logger = logging.getLogger(__name__)
 
@@ -243,7 +242,7 @@ async def _create_stack(
 
         plugins = [
             self._stats_plugin_factory.create_counter_plugin("Initial"),
-            BlacklistProviderIdPlugin(PROVIDERS_BLACKLIST.get(payment_network, set())),
+            ProviderBlacklistPlugin(payment_network),
             self._stats_plugin_factory.create_counter_plugin("Not blacklisted"),
         ]
 

ray_on_golem/reputation/__init__.py

@@ -0,0 +1,3 @@
+from ray_on_golem.reputation.service import ReputationService
+
+__all__ = ("ReputationService",)

ray_on_golem/reputation/admin.py

@@ -0,0 +1,109 @@
+import asyncio
+
+import click
+from aerich import DowngradeError, Migrate
+
+from ray_on_golem.cli import with_datadir
+from ray_on_golem.reputation.service import ReputationService
+
+
+def _format_migrations(migrations):
+    return [f"    {m}\n" for m in migrations]
+
+
+@click.group(help="Reputation subsystem admin.", context_settings={"show_default": True})
+def admin():
+    ...
+
+
+@admin.command(help="Generate a new migration file.")
+@with_datadir
+@click.option(
+    "--name",
+    type=str,
+    help="The name of the generated migration file",
+)
+@click.option(
+    "--empty",
+    is_flag=True,
+    help="If set, just generate an empty migrations file",
+)
+def migrate(datadir, name, empty):
+    if empty and not name:
+        print("Name is required for an empty migration.")
+        return
+
+    async def _migrate():
+        async with ReputationService(datadir, auto_apply_migrations=False) as service:
+            await service.migrations.init()
+            if empty:
+                migration_name = await Migrate._generate_diff_py(*([name] if name else []))  # noqa
+            else:
+                migration_name = await service.migrations.migrate(*([name] if name else []))
+
+        if migration_name:
+            print("Created a migration: ", migration_name)
+
+    asyncio.run(_migrate())
+
+
+@admin.command(help="Show the DB migration files.")
+@with_datadir
+@click.option(
+    "-f",
+    "--show-full",
+    is_flag=True,
+    help="If set, just generate an empty migrations file",
+)
+def updates(datadir, show_full):
+    async def _updates():
+        async with ReputationService(datadir, auto_apply_migrations=False) as service:
+            db_version = await Migrate.get_last_version()
+            if not db_version:
+                print("DB uninitialized.")
+                return
+
+            print("Current DB version: ", db_version.version, "\n")
+
+            migrations_available = await service.migrations_available()
+            if migrations_available:
+                print("Available updates: \n", *_format_migrations(migrations_available))
+            else:
+                print("The DB seems to be up to date :)")
+            if show_full:
+                print(
+                    "Full migration history: \n",
+                    *_format_migrations(await service.migrations.history()),
+                )
+
+    asyncio.run(_updates())
+
+
+@admin.command(help="Run all the updates.")
+@with_datadir
+def upgrade(datadir):
+    async def _upgrade():
+        async with ReputationService(datadir, auto_apply_migrations=False) as service:
+            applied_migrations = await service.apply_migrations()
+            if applied_migrations:
+                print("Applied migrations: \n", *_format_migrations(applied_migrations))
+            else:
+                print("No new migrations found.")
+
+    asyncio.run(_upgrade())
+
+
+@admin.command(help="Rollback all the migrations up to and including the given version.")
+@with_datadir
+@click.argument("version", type=int)
+def rollback(datadir, version: int):
+    async def _downgrade():
+        async with ReputationService(datadir, auto_apply_migrations=False) as service:
+            try:
+                rolled_back_migrations = await service.downgrade_migrations(version)
+                if rolled_back_migrations:
+                    print("Rolled-back migrations: \n", *_format_migrations(rolled_back_migrations))
+            except DowngradeError:
+                print("Migration not found in the DB.")
+
+    asyncio.run(_downgrade())