fix: Use fixed event dispatcher node ID (#513)

* fix: Use hostname or explicitly configured node ID as event dispatcher node ID
* setup: Upgrade aiotools to 1.4.0

Backported-From: main (22.03)
Backported-To: 21.03

changes/513.fix.md

@@ -0,0 +1,2 @@
+Use a fixed value as the node ID in `EventDispatcher` instances, either auto-generated from the hostname or manually configured `manager.id` value of `manager.toml`.
+- **IMPORTANT: An explicit admin/developer action is required** to fix up the corrupted Redis database and configuration. Check out the description of [lablup/backend.ai-manager#513](https://github.com/lablup/backend.ai-manager/pull/513) for details.

config/sample.toml

@@ -58,6 +58,8 @@ heartbeat-timeout = 30.0
 # Override the name of this manager node.
 # If empty or unspecified, the agent builds this from the hostname by prefixing it with "i-",
 # like "i-hostname".  The "i-" prefix is not mandatory, though.
+# Explicit configuration may be required if the hostname changes frequently,
+# to handle the event bus messages consistently.
 # This affects the per-node configuration scope.
 # id = ""
 

setup.cfg

@@ -43,7 +43,7 @@ install_requires =
     aiojobs~=0.3.0
     aiomonitor~=0.4.5
     aioredis[hiredis]~=2.0
-    aiotools~=1.2.2
+    aiotools~=1.4.0
     alembic~=1.6.5
     async_timeout~=3.0
     asyncache>=0.1.1

src/ai/backend/manager/config.py

@@ -205,6 +205,7 @@
 from pathlib import Path
 from pprint import pformat
 import secrets
+import socket
 import sys
 from typing import (
     Any,
@@ -282,6 +283,8 @@
     }),
     t.Key('manager'): t.Dict({
         t.Key('num-proc', default=_max_cpu_count): t.Int[1:_max_cpu_count],
+        t.Key('id', default=f"i-{socket.gethostname()}"): t.String,
+        t.Key('user', default=None): tx.UserID(default_uid=_file_perm.st_uid),
         t.Key('user', default=None): tx.UserID(default_uid=_file_perm.st_uid),
         t.Key('group', default=None): tx.GroupID(default_gid=_file_perm.st_gid),
         t.Key('service-addr', default=('0.0.0.0', 8080)): tx.HostPortPair,

src/ai/backend/manager/server.py

@@ -324,6 +324,7 @@ async def event_dispatcher_ctx(root_ctx: RootContext) -> AsyncIterator[None]:
         root_ctx.shared_config.data['redis'],
         db=REDIS_STREAM_DB,
         log_events=root_ctx.local_config['debug']['log-events'],
+        node_id=root_ctx.local_config['manager']['id'],
     )
     yield
     await root_ctx.event_dispatcher.close()

tests/test_distributed.py

@@ -85,6 +85,7 @@ async def _tick(context: Any, source: AgentId, event: NoopEvent) -> None:
         event_dispatcher = await EventDispatcher.new(
             self.shared_config.data['redis'],
             db=REDIS_STREAM_DB,
+            node_id=self.local_config['manager']['id'],
         )
         event_producer = await EventProducer.new(
             self.shared_config.data['redis'],
@@ -161,6 +162,7 @@ async def _tick(context: Any, source: AgentId, event: NoopEvent) -> None:
     event_dispatcher = await EventDispatcher.new(
         shared_config.data['redis'],
         db=REDIS_STREAM_DB,
+        node_id=local_config['manager']['id'],
     )
     event_producer = await EventProducer.new(
         shared_config.data['redis'],