Ensures client and worker IDs are always unique (fixes finos#419). (finos#420)

Signed-off-by: rafa-be <raphael@noisycamp.com>

Author: rafa-be

src/scaler/cluster/combo.py

@@ -89,7 +89,7 @@ def __init__(
             object_storage_address=self._object_storage_address,
             preload=None,
             worker_io_threads=worker_io_threads,
-            worker_names=[f"{socket.gethostname().split('.')[0]}_{i}" for i in range(n_workers)],
+            worker_names=[f"{socket.gethostname().split('.')[0]}" for _ in range(n_workers)],
             per_worker_capabilities=per_worker_capabilities or {},
             per_worker_task_queue_size=per_worker_task_queue_size,
             heartbeat_interval_seconds=heartbeat_interval_seconds,

src/scaler/utility/identifiers.py

@@ -16,10 +16,12 @@ def __repr__(self) -> str:
 
     @staticmethod
     def generate_client_id(name: Optional[str] = None) -> "ClientID":
-        if name is None:
-            name = uuid.uuid4().bytes.hex()
+        unique_client_tag = uuid.uuid4().bytes.hex()
 
-        return ClientID(f"Client|{name}".encode())
+        if name is None:
+            return ClientID(f"Client|{unique_client_tag}".encode())
+        else:
+            return ClientID(f"Client|{name}|{unique_client_tag}".encode())
 
 
 class WorkerID(Identifier):
@@ -35,7 +37,8 @@ def invalid_worker_id() -> "WorkerID":
 
     @staticmethod
     def generate_worker_id(name: str) -> "WorkerID":
-        return WorkerID(f"Worker|{name}".encode())
+        unique_worker_tag = uuid.uuid4().bytes.hex()
+        return WorkerID(f"Worker|{name}|{unique_worker_tag}".encode())
 
 
 _INVALID_WORKER_ID = WorkerID(b"")