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/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