feat: add node roles to del dag (PostHog#28155)

Co-authored-by: greptile-apps[bot] <165735046+greptile-apps[bot]@users.noreply.github.com>

dags/deletes.py

@@ -18,6 +18,7 @@
     ClickhouseCluster,
     Mutation,
     MutationRunner,
+    NodeRole,
     get_cluster,
 )
 from posthog.models.event.sql import EVENTS_DATA_TABLE
@@ -256,7 +257,7 @@ def delete_person_events(
     def sync_replica(client: Client):
         client.execute(f"SYSTEM SYNC REPLICA {load_pending_person_deletions.qualified_name} STRICT")
 
-    cluster.map_all_hosts(sync_replica).result()
+    cluster.map_hosts_by_role(sync_replica, NodeRole.WORKER).result()
 
     def count_pending_deletes(client: Client) -> int:
         result = client.execute(
@@ -267,7 +268,7 @@ def count_pending_deletes(client: Client) -> int:
         )
         return result[0][0] if result else 0
 
-    count_result = cluster.any_host(count_pending_deletes).result()
+    count_result = cluster.any_host_by_role(count_pending_deletes, NodeRole.WORKER).result()
 
     if count_result == 0:
         context.add_output_metadata({"events_deleted": MetadataValue.int(0), "message": "No pending deletions found"})

posthog/clickhouse/cluster.py

@@ -155,6 +155,17 @@ def any_host(self, fn: Callable[[Client], T]) -> Future[T]:
             host = self.__hosts[0]
             return executor.submit(self.__get_task_function(host, fn))
 
+    def any_host_by_role(self, fn: Callable[[Client], T], node_role: NodeRole) -> Future[T]:
+        """
+        Execute the callable once for any host with the given node role.
+        """
+        with ThreadPoolExecutor() as executor:
+            try:
+                host = next(host for host in self.__hosts if host.host_cluster_role == node_role.value.lower())
+            except StopIteration:
+                raise ValueError(f"No hosts found with role {node_role.value}")
+            return executor.submit(self.__get_task_function(host, fn))
+
     def map_all_hosts(self, fn: Callable[[Client], T], concurrency: int | None = None) -> FuturesMap[HostInfo, T]:
         """
         Execute the callable once for each host in the cluster.