feat: add workers autodicsovery feature (#121)

* add workers autodicsovery feature

* suppress erros while retrieving worker list

* update recipes

* fix tests and update docs

* filter health checks

* ping based workers

tests/ahriman/application/handlers/test_handler_validate.py

@@ -79,6 +79,7 @@ def test_schema(configuration: Configuration) -> None:
     assert schema.pop("s3")
     assert schema.pop("telegram")
     assert schema.pop("upload")
+    assert schema.pop("worker")
 
     assert schema == CONFIGURATION_SCHEMA
 

tests/ahriman/application/handlers/test_handler_web.py

@@ -39,6 +39,7 @@ def test_run(args: argparse.Namespace, configuration: Configuration, repository:
     setup_mock = mocker.patch("ahriman.web.web.setup_server")
     run_mock = mocker.patch("ahriman.web.web.run_server")
     start_mock = mocker.patch("ahriman.core.spawn.Spawn.start")
+    trigger_mock = mocker.patch("ahriman.core.triggers.TriggerLoader.load")
     stop_mock = mocker.patch("ahriman.core.spawn.Spawn.stop")
     join_mock = mocker.patch("ahriman.core.spawn.Spawn.join")
     _, repository_id = configuration.check_loaded()
@@ -48,6 +49,8 @@ def test_run(args: argparse.Namespace, configuration: Configuration, repository:
     setup_mock.assert_called_once_with(configuration, pytest.helpers.anyvar(int), [repository_id])
     run_mock.assert_called_once_with(pytest.helpers.anyvar(int))
     start_mock.assert_called_once_with()
+    trigger_mock.assert_called_once_with(repository_id, configuration)
+    trigger_mock().on_start.assert_called_once_with()
     stop_mock.assert_called_once_with()
     join_mock.assert_called_once_with()
 

tests/ahriman/core/distributed/conftest.py

@@ -0,0 +1,35 @@
+import pytest
+
+from ahriman.core.configuration import Configuration
+from ahriman.core.distributed import WorkersCache
+from ahriman.core.distributed.distributed_system import DistributedSystem
+
+
+@pytest.fixture
+def distributed_system(configuration: Configuration) -> DistributedSystem:
+    """
+    distributed system fixture
+
+    Args:
+        configuration(Configuration): configuration fixture
+
+    Returns:
+        DistributedSystem: distributed system test instance
+    """
+    configuration.set_option("status", "address", "http://localhost:8081")
+    _, repository_id = configuration.check_loaded()
+    return DistributedSystem(repository_id, configuration)
+
+
+@pytest.fixture
+def workers_cache(configuration: Configuration) -> WorkersCache:
+    """
+    workers cache fixture
+
+    Args:
+        configuration(Configuration): configuration fixture
+
+    Returns:
+        WorkersCache: workers cache test instance
+    """
+    return WorkersCache(configuration)

tests/ahriman/core/distributed/test_distributed_system.py

@@ -0,0 +1,82 @@
+import json
+import requests
+
+from pytest_mock import MockerFixture
+
+from ahriman.core.configuration import Configuration
+from ahriman.core.distributed.distributed_system import DistributedSystem
+from ahriman.models.worker import Worker
+
+
+def test_configuration_sections(configuration: Configuration) -> None:
+    """
+    must correctly parse target list
+    """
+    assert DistributedSystem.configuration_sections(configuration) == ["worker"]
+
+
+def test_workers_url(distributed_system: DistributedSystem) -> None:
+    """
+    must generate workers url correctly
+    """
+    assert distributed_system._workers_url().startswith(distributed_system.address)
+    assert distributed_system._workers_url().endswith("/api/v1/distributed")
+
+
+def test_register(distributed_system: DistributedSystem, mocker: MockerFixture) -> None:
+    """
+    must register service
+    """
+    run_mock = mocker.patch("ahriman.core.distributed.distributed_system.DistributedSystem.make_request")
+    distributed_system.register()
+    run_mock.assert_called_once_with("POST", f"{distributed_system.address}/api/v1/distributed",
+                                     json=distributed_system.worker.view())
+
+
+def test_register_failed(distributed_system: DistributedSystem, mocker: MockerFixture) -> None:
+    """
+    must suppress any exception happened during worker registration
+    """
+    mocker.patch("requests.Session.request", side_effect=Exception())
+    distributed_system.register()
+
+
+def test_register_failed_http_error(distributed_system: DistributedSystem, mocker: MockerFixture) -> None:
+    """
+    must suppress HTTP exception happened during worker registration
+    """
+    mocker.patch("requests.Session.request", side_effect=requests.HTTPError())
+    distributed_system.register()
+
+
+def test_workers(distributed_system: DistributedSystem, mocker: MockerFixture) -> None:
+    """
+    must return available remote workers
+    """
+    worker = Worker("remote")
+    response_obj = requests.Response()
+    response_obj._content = json.dumps([worker.view()]).encode("utf8")
+    response_obj.status_code = 200
+
+    requests_mock = mocker.patch("ahriman.core.status.web_client.WebClient.make_request",
+                                 return_value=response_obj)
+
+    result = distributed_system.workers()
+    requests_mock.assert_called_once_with("GET", distributed_system._workers_url())
+    assert result == [worker]
+
+
+def test_workers_failed(distributed_system: DistributedSystem, mocker: MockerFixture) -> None:
+    """
+    must suppress any exception happened during worker extraction
+    """
+    mocker.patch("requests.Session.request", side_effect=Exception())
+    distributed_system.workers()
+
+
+def test_workers_failed_http_error(distributed_system: DistributedSystem, mocker: MockerFixture) -> None:
+    """
+    must suppress HTTP exception happened during worker extraction
+    """
+    mocker.patch("requests.Session.request", side_effect=requests.HTTPError())
+    distributed_system.workers()

tests/ahriman/core/distributed/test_worker_loader_trigger.py

@@ -0,0 +1,47 @@
+from pytest_mock import MockerFixture
+
+from ahriman.core.configuration import Configuration
+from ahriman.core.distributed import WorkerLoaderTrigger
+from ahriman.models.worker import Worker
+
+
+def test_on_start(configuration: Configuration, mocker: MockerFixture) -> None:
+    """
+    must load workers from remote
+    """
+    worker = Worker("address")
+    configuration.set_option("status", "address", "http://localhost:8081")
+    run_mock = mocker.patch("ahriman.core.distributed.WorkerLoaderTrigger.workers", return_value=[worker])
+    _, repository_id = configuration.check_loaded()
+
+    trigger = WorkerLoaderTrigger(repository_id, configuration)
+    trigger.on_start()
+    run_mock.assert_called_once_with()
+    assert configuration.getlist("build", "workers") == [worker.address]
+
+
+def test_on_start_skip(configuration: Configuration, mocker: MockerFixture) -> None:
+    """
+    must skip loading if option is already set
+    """
+    configuration.set_option("status", "address", "http://localhost:8081")
+    configuration.set_option("build", "workers", "address")
+    run_mock = mocker.patch("ahriman.core.distributed.WorkerLoaderTrigger.workers")
+    _, repository_id = configuration.check_loaded()
+
+    trigger = WorkerLoaderTrigger(repository_id, configuration)
+    trigger.on_start()
+    run_mock.assert_not_called()
+
+
+def test_on_start_empty_list(configuration: Configuration, mocker: MockerFixture) -> None:
+    """
+    must do not set anything if workers are not available
+    """
+    configuration.set_option("status", "address", "http://localhost:8081")
+    mocker.patch("ahriman.core.distributed.WorkerLoaderTrigger.workers", return_value=[])
+    _, repository_id = configuration.check_loaded()
+
+    trigger = WorkerLoaderTrigger(repository_id, configuration)
+    trigger.on_start()
+    assert not configuration.has_option("build", "workers")

tests/ahriman/core/distributed/test_worker_trigger.py

@@ -0,0 +1,52 @@
+from pytest_mock import MockerFixture
+
+from ahriman.core.configuration import Configuration
+from ahriman.core.distributed import WorkerTrigger
+
+
+def test_on_start(configuration: Configuration, mocker: MockerFixture) -> None:
+    """
+    must register itself as worker
+    """
+    configuration.set_option("status", "address", "http://localhost:8081")
+    run_mock = mocker.patch("threading.Timer.start")
+    _, repository_id = configuration.check_loaded()
+
+    WorkerTrigger(repository_id, configuration).on_start()
+    run_mock.assert_called_once_with()
+
+
+def test_on_stop(configuration: Configuration, mocker: MockerFixture) -> None:
+    """
+    must unregister itself as worker
+    """
+    configuration.set_option("status", "address", "http://localhost:8081")
+    run_mock = mocker.patch("threading.Timer.cancel")
+    _, repository_id = configuration.check_loaded()
+
+    WorkerTrigger(repository_id, configuration).on_stop()
+    run_mock.assert_called_once_with()
+
+
+def test_on_stop_empty_timer(configuration: Configuration) -> None:
+    """
+    must do not fail if no timer was started
+    """
+    configuration.set_option("status", "address", "http://localhost:8081")
+    _, repository_id = configuration.check_loaded()
+
+    WorkerTrigger(repository_id, configuration).on_stop()
+
+
+def test_ping(configuration: Configuration, mocker: MockerFixture) -> None:
+    """
+    must correctly process timer action
+    """
+    configuration.set_option("status", "address", "http://localhost:8081")
+    run_mock = mocker.patch("ahriman.core.distributed.WorkerTrigger.register")
+    timer_mock = mocker.patch("threading.Timer.start")
+    _, repository_id = configuration.check_loaded()
+
+    WorkerTrigger(repository_id, configuration).ping()
+    run_mock.assert_called_once_with()
+    timer_mock.assert_called_once_with()

tests/ahriman/core/distributed/test_workers_cache.py

@@ -0,0 +1,43 @@
+import time
+
+from ahriman.core.distributed import WorkersCache
+from ahriman.models.worker import Worker
+
+
+def test_workers(workers_cache: WorkersCache) -> None:
+    """
+    must return alive workers
+    """
+    workers_cache._workers = {
+        str(index): (Worker(f"address{index}"), index)
+        for index in range(2)
+    }
+    workers_cache.time_to_live = time.monotonic()
+
+    assert workers_cache.workers == [Worker("address1")]
+
+
+def test_workers_remove(workers_cache: WorkersCache) -> None:
+    """
+    must remove all workers
+    """
+    workers_cache.workers_update(Worker("address"))
+    assert workers_cache.workers
+
+    workers_cache.workers_remove()
+    assert not workers_cache.workers
+
+
+def test_workers_update(workers_cache: WorkersCache) -> None:
+    """
+    must update worker
+    """
+    worker = Worker("address")
+
+    workers_cache.workers_update(worker)
+    assert workers_cache.workers == [worker]
+    _, first_last_seen = workers_cache._workers[worker.identifier]
+
+    workers_cache.workers_update(worker)
+    _, second_last_seen = workers_cache._workers[worker.identifier]
+    assert first_last_seen < second_last_seen

tests/ahriman/core/log/test_filtered_access_logger.py

@@ -4,6 +4,44 @@ from unittest.mock import MagicMock
 from ahriman.core.log.filtered_access_logger import FilteredAccessLogger
 
 
+def test_is_distributed_post() -> None:
+    """
+    must correctly define distributed services ping request
+    """
+    request = MagicMock()
+
+    request.method = "POST"
+    request.path = "/api/v1/distributed"
+    assert FilteredAccessLogger.is_distributed_post(request)
+
+    request.method = "GET"
+    request.path = "/api/v1/distributed"
+    assert not FilteredAccessLogger.is_distributed_post(request)
+
+    request.method = "POST"
+    request.path = "/api/v1/distributed/path"
+    assert not FilteredAccessLogger.is_distributed_post(request)
+
+
+def test_is_info_get() -> None:
+    """
+    must correctly define health check request
+    """
+    request = MagicMock()
+
+    request.method = "GET"
+    request.path = "/api/v1/info"
+    assert FilteredAccessLogger.is_info_get(request)
+
+    request.method = "POST"
+    request.path = "/api/v1/info"
+    assert not FilteredAccessLogger.is_info_get(request)
+
+    request.method = "GET"
+    request.path = "/api/v1/infos"
+    assert not FilteredAccessLogger.is_info_get(request)
+
+
 def test_is_logs_post() -> None:
     """
     must correctly define if request belongs to logs posting

tests/ahriman/core/status/test_web_client.py

@@ -12,6 +12,7 @@ from ahriman.models.changes import Changes
 from ahriman.models.internal_status import InternalStatus
 from ahriman.models.log_record_id import LogRecordId
 from ahriman.models.package import Package
+from ahriman.models.worker import Worker
 
 
 def test_parse_address(configuration: Configuration) -> None:
@@ -32,14 +33,6 @@ def test_parse_address(configuration: Configuration) -> None:
     assert WebClient.parse_address(configuration) == ("status", "http://localhost:8082")
 
 
-def test_status_url(web_client: WebClient) -> None:
-    """
-    must generate package status url correctly
-    """
-    assert web_client._status_url().startswith(web_client.address)
-    assert web_client._status_url().endswith("/api/v1/status")
-
-
 def test_changes_url(web_client: WebClient, package_ahriman: Package) -> None:
     """
     must generate changes url correctly
@@ -67,6 +60,14 @@ def test_package_url(web_client: WebClient, package_ahriman: Package) -> None:
     assert web_client._package_url(package_ahriman.base).endswith(f"/api/v1/packages/{package_ahriman.base}")
 
 
+def test_status_url(web_client: WebClient) -> None:
+    """
+    must generate package status url correctly
+    """
+    assert web_client._status_url().startswith(web_client.address)
+    assert web_client._status_url().endswith("/api/v1/status")
+
+
 def test_package_add(web_client: WebClient, package_ahriman: Package, mocker: MockerFixture) -> None:
     """
     must process package addition

tests/ahriman/models/test_worker.py

@@ -8,3 +8,17 @@ def test_post_init() -> None:
     assert Worker("http://localhost:8080").identifier == "localhost:8080"
     assert Worker("remote").identifier == ""  # not a valid url
     assert Worker("remote", identifier="id").identifier == "id"
+
+
+def test_view() -> None:
+    """
+    must generate json view
+    """
+    worker = Worker("address")
+    assert worker.view() == {"address": worker.address, "identifier": worker.identifier}
+
+    worker = Worker("http://localhost:8080")
+    assert worker.view() == {"address": worker.address, "identifier": worker.identifier}
+
+    worker = Worker("http://localhost:8080", identifier="abc")
+    assert worker.view() == {"address": worker.address, "identifier": worker.identifier}

tests/ahriman/web/schemas/test_worker_schema.py

@@ -0,0 +1 @@
+# schema testing goes in view class tests

tests/ahriman/web/views/test_view_base.py

@@ -50,11 +50,18 @@ def test_spawn(base: BaseView) -> None:
 
 def test_validator(base: BaseView) -> None:
     """
-    must return service
+    must return validator service
     """
     assert base.validator
 
 
+def test_workers(base: BaseView) -> None:
+    """
+    must return worker service
+    """
+    assert base.workers
+
+
 async def test_get_permission(base: BaseView) -> None:
     """
     must search for permission attribute in class

tests/ahriman/web/views/v1/distributed/test_view_v1_distributed_workers.py

@@ -0,0 +1,80 @@
+import pytest
+
+from aiohttp.test_utils import TestClient
+
+from ahriman.models.user_access import UserAccess
+from ahriman.models.worker import Worker
+from ahriman.web.views.v1.distributed.workers import WorkersView
+
+
+async def test_get_permission() -> None:
+    """
+    must return correct permission for the request
+    """
+    for method in ("DELETE", "GET", "POST"):
+        request = pytest.helpers.request("", "", method)
+        assert await WorkersView.get_permission(request) == UserAccess.Full
+
+
+def test_routes() -> None:
+    """
+    must return correct routes
+    """
+    assert WorkersView.ROUTES == ["/api/v1/distributed"]
+
+
+async def test_delete(client: TestClient) -> None:
+    """
+    must delete all workers
+    """
+    await client.post("/api/v1/distributed", json={"address": "address1", "identifier": "1"})
+    await client.post("/api/v1/distributed", json={"address": "address2", "identifier": "2"})
+
+    response = await client.delete("/api/v1/distributed")
+    assert response.status == 204
+
+    response = await client.get("/api/v1/distributed")
+    json = await response.json()
+    assert not json
+
+
+async def test_get(client: TestClient) -> None:
+    """
+    must return all workers
+    """
+    await client.post("/api/v1/distributed", json={"address": "address1", "identifier": "1"})
+    await client.post("/api/v1/distributed", json={"address": "address2", "identifier": "2"})
+    response_schema = pytest.helpers.schema_response(WorkersView.get)
+
+    response = await client.get("/api/v1/distributed")
+    assert response.ok
+    json = await response.json()
+    assert not response_schema.validate(json, many=True)
+
+    workers = [Worker(item["address"], identifier=item["identifier"]) for item in json]
+    assert workers == [Worker("address1", identifier="1"), Worker("address2", identifier="2")]
+
+
+async def test_post(client: TestClient) -> None:
+    """
+    must update worker
+    """
+    worker = Worker("address1", identifier="1")
+    request_schema = pytest.helpers.schema_request(WorkersView.post)
+
+    payload = worker.view()
+    assert not request_schema.validate(payload)
+
+    response = await client.post("/api/v1/distributed", json=payload)
+    assert response.status == 204
+
+
+async def test_post_exception(client: TestClient) -> None:
+    """
+    must raise exception on invalid payload
+    """
+    response_schema = pytest.helpers.schema_response(WorkersView.post, code=400)
+
+    response = await client.post("/api/v1/distributed", json={})
+    assert response.status == 400
+    assert not response_schema.validate(await response.json())

tests/testresources/core/ahriman.ini

@@ -25,7 +25,7 @@ ignore_packages =
 makechrootpkg_flags =
 makepkg_flags = --skippgpcheck
 triggers = ahriman.core.report.ReportTrigger ahriman.core.upload.UploadTrigger
-triggers_known = ahriman.core.gitremote.RemotePullTrigger ahriman.core.gitremote.RemotePushTrigger ahriman.core.report.ReportTrigger ahriman.core.upload.UploadTrigger ahriman.core.support.KeyringTrigger ahriman.core.support.MirrorlistTrigger
+triggers_known = ahriman.core.distributed.WorkerLoaderTrigger ahriman.core.distributed.WorkerRegisterTrigger ahriman.core.distributed.WorkerTrigger ahriman.core.distributed.WorkerUnregisterTrigger ahriman.core.gitremote.RemotePullTrigger ahriman.core.gitremote.RemotePushTrigger ahriman.core.report.ReportTrigger ahriman.core.upload.UploadTrigger ahriman.core.support.KeyringTrigger ahriman.core.support.MirrorlistTrigger
 
 [repository]
 name = aur-clone
@@ -115,4 +115,7 @@ username = arcan1s
 enable_archive_upload = yes
 host = 127.0.0.1
 static_path = ../web/templates/static
-templates = ../web/templates
+templates = ../web/templates
+
+[worker]
+address = http://localhost:8081