From 4549510f2e1b3150c845681556b42c5a347a25dc Mon Sep 17 00:00:00 2001
From: Andrey Kholmogorov <63207701+Varagon007@users.noreply.github.com>
Date: Tue, 26 Sep 2023 11:59:13 +0200
Subject: [PATCH 1/7] Add external run action test-integration (#64)

* NEBIUSMDB-775 add ext run test_integration

* Correct wf name and inputs
---
 .github/workflows/test_clickhouse_version.yml | 27 +++++++++++++++++++
 1 file changed, 27 insertions(+)
 create mode 100644 .github/workflows/test_clickhouse_version.yml

diff --git a/.github/workflows/test_clickhouse_version.yml b/.github/workflows/test_clickhouse_version.yml
new file mode 100644
index 00000000..67b879a9
--- /dev/null
+++ b/.github/workflows/test_clickhouse_version.yml
@@ -0,0 +1,27 @@
+name: test_clickhouse_version
+
+run-name: ${{ github.workflow }}_${{ inputs.clickhouse_version }}_${{ inputs.id || github.run_number }}
+
+on:
+  workflow_dispatch:
+    inputs:
+      clickhouse_version:
+        description: 'ClickHouse version'
+        required: true
+        type: string
+      id:
+        description: 'Run identifier'
+        required: false
+        type: string
+        default: ""
+
+jobs:
+  test_integration:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+      - uses: ./.github/actions/setup_dependencies
+        with:
+          python-version: "3.11"
+      - name: run integration tests
+        run: CLICKHOUSE_VERSION=${{ inputs.clickhouse_version }} make test-integration

From ebec5a81e472e3ed0e31273b0e92c4e9fc9dca0e Mon Sep 17 00:00:00 2001
From: ianton-ru <56930273+ianton-ru@users.noreply.github.com>
Date: Wed, 27 Sep 2023 21:15:18 +0300
Subject: [PATCH 2/7] Add IPs in error message about external IPs (#66)

---
 ch_tools/monrun_checks/ext_ip_dns.py | 26 +++++++++++++++++---------
 1 file changed, 17 insertions(+), 9 deletions(-)

diff --git a/ch_tools/monrun_checks/ext_ip_dns.py b/ch_tools/monrun_checks/ext_ip_dns.py
index e4ee565f..aa6a0b41 100644
--- a/ch_tools/monrun_checks/ext_ip_dns.py
+++ b/ch_tools/monrun_checks/ext_ip_dns.py
@@ -1,7 +1,7 @@
 import json
 import socket
 from functools import lru_cache
-from typing import List
+from typing import List, Tuple
 
 import click
 import dns.resolver
@@ -56,7 +56,7 @@ def _check_fqdn(target: _TargetRecord, ipv6: bool, imdsv2: bool) -> list:
     err = []
     resolver = dns.resolver.Resolver()
 
-    def _compare(record_type: str, ip_type: str) -> bool:
+    def _compare(record_type: str, ip_type: str) -> Tuple[bool, set, set]:
         try:
             actual_addr = set(
                 map(lambda a: a.to_text(), resolver.resolve(target.fqdn, record_type))
@@ -65,13 +65,21 @@ def _compare(record_type: str, ip_type: str) -> bool:
             actual_addr = set()
         target_addr = {_get_host_ip(ip_type, imdsv2)}
         if target.strict:
-            return target_addr == actual_addr
-        return actual_addr.issuperset(target_addr)
-
-    if not _compare("A", "private_v4" if target.private else "public_v4"):
-        err.append(f"{target.fqdn}: invalid A")
-    if ipv6 and not _compare("AAAA", "ipv6"):
-        err.append(f"{target.fqdn}: invalid AAAA")
+            return target_addr == actual_addr, target_addr, actual_addr
+        return actual_addr.issuperset(target_addr), target_addr, actual_addr
+
+    ok, target_addr, actual_addr = _compare(
+        "A", "private_v4" if target.private else "public_v4"
+    )
+    if not ok:
+        err.append(
+            f"{target.fqdn}: invalid A: expected {target_addr}, actual {actual_addr}"
+        )
+    ok, target_addr, actual_addr = _compare("AAAA", "ipv6")
+    if ipv6 and not ok:
+        err.append(
+            f"{target.fqdn}: invalid AAAA: expected {target_addr}, actual {actual_addr}"
+        )
 
     return err
 

From 5f0488e3ab815cc7ef0e99b649fe60445c3475eb Mon Sep 17 00:00:00 2001
From: Alexander Burmak <Alex-Burmak@yandex-team.ru>
Date: Thu, 28 Sep 2023 11:18:25 -0700
Subject: [PATCH 3/7] Fixed handling of --on-cluster option in "table delete"
 and "database delete" commands (#65)

---
 ch_tools/chadmin/cli/database_group.py | 15 +++++++++++++--
 ch_tools/chadmin/cli/table_group.py    | 21 +++++++++++++++++++--
 ch_tools/chadmin/internal/table.py     |  7 ++++++-
 3 files changed, 38 insertions(+), 5 deletions(-)

diff --git a/ch_tools/chadmin/cli/database_group.py b/ch_tools/chadmin/cli/database_group.py
index b22c4f77..5adb4d52 100644
--- a/ch_tools/chadmin/cli/database_group.py
+++ b/ch_tools/chadmin/cli/database_group.py
@@ -1,5 +1,6 @@
 from click import argument, group, option, pass_context
 
+from ch_tools.chadmin.cli import get_cluster_name
 from ch_tools.chadmin.internal.utils import execute_query
 
 
@@ -47,7 +48,13 @@ def list_databases_command(ctx, **kwargs):
 @option("-d", "--database")
 @option("--exclude-database")
 @option("-a", "--all", "all_", is_flag=True, help="Delete all databases.")
-@option("--cluster")
+@option(
+    "--cluster",
+    "--on-cluster",
+    "on_cluster",
+    is_flag=True,
+    help="Delete databases on all hosts of the cluster.",
+)
 @option(
     "-n",
     "--dry-run",
@@ -55,10 +62,14 @@ def list_databases_command(ctx, **kwargs):
     default=False,
     help="Enable dry run mode and do not perform any modifying actions.",
 )
-def delete_databases_command(ctx, dry_run, all_, database, exclude_database, cluster):
+def delete_databases_command(
+    ctx, dry_run, all_, database, exclude_database, on_cluster
+):
     if not any((all_, database)):
         ctx.fail("At least one of --all, --database options must be specified.")
 
+    cluster = get_cluster_name(ctx) if on_cluster else None
+
     for d in get_databases(
         ctx, database=database, exclude_database=exclude_database, format_="JSON"
     )["data"]:
diff --git a/ch_tools/chadmin/cli/table_group.py b/ch_tools/chadmin/cli/table_group.py
index 639be479..7d684b2d 100644
--- a/ch_tools/chadmin/cli/table_group.py
+++ b/ch_tools/chadmin/cli/table_group.py
@@ -120,7 +120,19 @@ def columns_command(ctx, database, table):
     "--exclude-table", help="Filter out tables to delete by the specified table name."
 )
 @option("-a", "--all", "all_", is_flag=True, help="Delete all tables.")
-@option("--cluster")
+@option(
+    "--cluster",
+    "--on-cluster",
+    "on_cluster",
+    is_flag=True,
+    help="Delete tables on all hosts of the cluster.",
+)
+@option(
+    "--sync/--async",
+    "sync_mode",
+    default=True,
+    help="Enable/Disable synchronous query execution.",
+)
 @option(
     "-n",
     "--dry-run",
@@ -128,7 +140,9 @@ def columns_command(ctx, database, table):
     default=False,
     help="Enable dry run mode and do not perform any modifying actions.",
 )
-def delete_command(ctx, dry_run, all_, database, table, exclude_table, cluster):
+def delete_command(
+    ctx, all_, database, table, exclude_table, on_cluster, sync_mode, dry_run
+):
     """
     Delete one or several tables.
     """
@@ -137,6 +151,8 @@ def delete_command(ctx, dry_run, all_, database, table, exclude_table, cluster):
             "At least one of --all, --database, --table options must be specified."
         )
 
+    cluster = get_cluster_name(ctx) if on_cluster else None
+
     for t in list_tables(
         ctx, database=database, table=table, exclude_table=exclude_table
     ):
@@ -145,6 +161,7 @@ def delete_command(ctx, dry_run, all_, database, table, exclude_table, cluster):
             database=t["database"],
             table=t["table"],
             cluster=cluster,
+            sync_mode=sync_mode,
             echo=True,
             dry_run=dry_run,
         )
diff --git a/ch_tools/chadmin/internal/table.py b/ch_tools/chadmin/internal/table.py
index 86567fd2..b38b6b77 100644
--- a/ch_tools/chadmin/internal/table.py
+++ b/ch_tools/chadmin/internal/table.py
@@ -149,7 +149,9 @@ def attach_table(ctx, database, table, *, cluster=None, echo=False, dry_run=Fals
     )
 
 
-def delete_table(ctx, database, table, *, cluster=None, echo=False, dry_run=False):
+def delete_table(
+    ctx, database, table, *, cluster=None, echo=False, sync_mode=True, dry_run=False
+):
     """
     Perform "DROP TABLE" for the specified table.
     """
@@ -158,7 +160,9 @@ def delete_table(ctx, database, table, *, cluster=None, echo=False, dry_run=Fals
         {%- if cluster %}
         ON CLUSTER '{{ cluster }}'
         {%- endif %}
+        {%- if sync_mode %}
         NO DELAY
+        {%- endif %}
         """
     execute_query(
         ctx,
@@ -166,6 +170,7 @@ def delete_table(ctx, database, table, *, cluster=None, echo=False, dry_run=Fals
         database=database,
         table=table,
         cluster=cluster,
+        sync_mode=sync_mode,
         echo=echo,
         dry_run=dry_run,
         format_=None,

From 364f5f0d130762b06ddb7834fd77fa9a2f30a4e6 Mon Sep 17 00:00:00 2001
From: Alexander Burmak <Alex-Burmak@yandex-team.ru>
Date: Thu, 28 Sep 2023 11:22:07 -0700
Subject: [PATCH 4/7] ClickHouse: --limit option for 'mutation list' command
 (#67)

---
 ch_tools/chadmin/cli/mutation_group.py | 9 ++++++++-
 1 file changed, 8 insertions(+), 1 deletion(-)

diff --git a/ch_tools/chadmin/cli/mutation_group.py b/ch_tools/chadmin/cli/mutation_group.py
index 060acfea..79208e1c 100644
--- a/ch_tools/chadmin/cli/mutation_group.py
+++ b/ch_tools/chadmin/cli/mutation_group.py
@@ -53,8 +53,11 @@ def get_mutation(ctx, mutation, last):
     is_flag=True,
     help="Get mutations from all hosts in the cluster.",
 )
+@option(
+    "-l", "--limit", type=int, help="Limit the max number of objects in the output."
+)
 @pass_context
-def list_mutations(ctx, is_done, command_pattern, on_cluster):
+def list_mutations(ctx, is_done, command_pattern, on_cluster, limit):
     """List mutations."""
     cluster = get_cluster_name(ctx) if on_cluster else None
     query = """
@@ -86,6 +89,9 @@ def list_mutations(ctx, is_done, command_pattern, on_cluster):
         {% if command_pattern %}
           AND command ILIKE '{{ command_pattern }}'
         {% endif %}
+        {% if limit -%}
+        LIMIT {{ limit }}
+        {% endif -%}
         """
     response = execute_query(
         ctx,
@@ -93,6 +99,7 @@ def list_mutations(ctx, is_done, command_pattern, on_cluster):
         is_done=is_done,
         command_pattern=command_pattern,
         cluster=cluster,
+        limit=limit,
         format_="Vertical",
     )
     print(response)

From bf5385c44ee56e038aa6006f380658d56a6f8b6a Mon Sep 17 00:00:00 2001
From: Andrey Kholmogorov <63207701+Varagon007@users.noreply.github.com>
Date: Fri, 29 Sep 2023 16:47:05 +0200
Subject: [PATCH 5/7] Add option step for create test-image (#68)

---
 .github/workflows/test_clickhouse_version.yml | 22 ++++++++++++++++++-
 1 file changed, 21 insertions(+), 1 deletion(-)

diff --git a/.github/workflows/test_clickhouse_version.yml b/.github/workflows/test_clickhouse_version.yml
index 67b879a9..016dfefe 100644
--- a/.github/workflows/test_clickhouse_version.yml
+++ b/.github/workflows/test_clickhouse_version.yml
@@ -18,10 +18,30 @@ on:
 jobs:
   test_integration:
     runs-on: ubuntu-latest
+    env:
+      CLICKHOUSE_VERSION: ${{ inputs.clickhouse_version }}
+      NEED_SETUP: true
     steps:
       - uses: actions/checkout@v3
+      - name: Check the input tag to ${{ inputs.clickhouse_version }}
+        continue-on-error: true
+        run: |
+          docker manifest inspect chtools/test-clickhouse:${{ inputs.clickhouse_version }}
+          echo "NEED_SETUP=false" >> $GITHUB_ENV
+      - name: login to dockerhub
+        if: env.NEED_SETUP == 'true'
+        uses: docker/login-action@v2
+        with:
+          username: ${{ secrets.DOCKERHUB_USERNAME }}
+          password: ${{ secrets.DOCKERHUB_TOKEN }}
+      - name: build and push necessary images
+        if: env.NEED_SETUP == 'true'
+        uses: docker/bake-action@v3
+        with:
+          files: tests/bake.hcl
+          push: true
       - uses: ./.github/actions/setup_dependencies
         with:
           python-version: "3.11"
       - name: run integration tests
-        run: CLICKHOUSE_VERSION=${{ inputs.clickhouse_version }} make test-integration
+        run:  make test-integration

From 3024d65ec5f4fe04ce54f71b56f48310715f8b1e Mon Sep 17 00:00:00 2001
From: Andrey Kholmogorov <63207701+Varagon007@users.noreply.github.com>
Date: Fri, 29 Sep 2023 17:45:29 +0200
Subject: [PATCH 6/7] NEBIUSMDB-777 add env for push inage (#69)

---
 .github/workflows/test_clickhouse_version.yml | 1 +
 1 file changed, 1 insertion(+)

diff --git a/.github/workflows/test_clickhouse_version.yml b/.github/workflows/test_clickhouse_version.yml
index 016dfefe..3207ef5b 100644
--- a/.github/workflows/test_clickhouse_version.yml
+++ b/.github/workflows/test_clickhouse_version.yml
@@ -20,6 +20,7 @@ jobs:
     runs-on: ubuntu-latest
     env:
       CLICKHOUSE_VERSION: ${{ inputs.clickhouse_version }}
+      CLICKHOUSE_VERSIONS: ${{ inputs.clickhouse_version }}
       NEED_SETUP: true
     steps:
       - uses: actions/checkout@v3

From 8c0735165cd65d296db494c1cb1eee342946c1f9 Mon Sep 17 00:00:00 2001
From: MikhailBurdukov <102754618+MikhailBurdukov@users.noreply.github.com>
Date: Thu, 5 Oct 2023 12:59:28 +0300
Subject: [PATCH 7/7] Move zookeeper.py in chadmin (#70)

* Move zookeeper.py in chadmin

* Review fixes

* Changes

* Update chadmin_zookeeper.feature
---
 ch_tools/chadmin/cli/zookeeper_group.py  |  24 +++-
 ch_tools/chadmin/internal/zookeeper.py   | 168 +++++++++++++++++++++--
 tests/features/chadmin_zookeeper.feature |  43 ++++++
 tests/modules/chadmin.py                 |  35 +++++
 tests/steps/chadmin.py                   |  40 ++++++
 5 files changed, 296 insertions(+), 14 deletions(-)
 create mode 100644 tests/features/chadmin_zookeeper.feature
 create mode 100644 tests/modules/chadmin.py
 create mode 100644 tests/steps/chadmin.py

diff --git a/ch_tools/chadmin/cli/zookeeper_group.py b/ch_tools/chadmin/cli/zookeeper_group.py
index f50ec367..154106c5 100644
--- a/ch_tools/chadmin/cli/zookeeper_group.py
+++ b/ch_tools/chadmin/cli/zookeeper_group.py
@@ -7,6 +7,7 @@
 from ch_tools.chadmin.internal.table_replica import get_table_replica
 from ch_tools.chadmin.internal.zookeeper import (
     check_zk_node,
+    clean_zk_metadata_for_hosts,
     create_zk_nodes,
     delete_zk_nodes,
     get_zk_node,
@@ -41,8 +42,18 @@
     help="Do not try to get parameters from clickhouse config.xml.",
     default=False,
 )
+@option(
+    "-c",
+    "--chroot",
+    "zk_root_path",
+    type=str,
+    help="Cluster ZooKeeper root path. If not specified,the root path will be used.",
+    required=False,
+)
 @pass_context
-def zookeeper_group(ctx, host, port, timeout, zkcli_identity, no_chroot, no_ch_config):
+def zookeeper_group(
+    ctx, host, port, timeout, zkcli_identity, no_chroot, no_ch_config, zk_root_path
+):
     """ZooKeeper management commands.
 
     ZooKeeper command runs client which connects to Zookeeper node.
@@ -57,6 +68,7 @@ def zookeeper_group(ctx, host, port, timeout, zkcli_identity, no_chroot, no_ch_c
         "zkcli_identity": zkcli_identity,
         "no_chroot": no_chroot,
         "no_ch_config": no_ch_config,
+        "zk_root_path": zk_root_path,
     }
 
 
@@ -266,3 +278,13 @@ def delete_ddl_task_command(ctx, tasks):
     """
     paths = [f"/clickhouse/task_queue/ddl/{task}" for task in tasks]
     delete_zk_nodes(ctx, paths)
+
+
+@zookeeper_group.command(
+    name="cleanup-removed-hosts-metadata",
+    help="Remove metadata from Zookeeper for specified hosts.",
+)
+@argument("fqdn", type=ListParamType())
+@pass_context
+def clickhouse_hosts_command(ctx, fqdn):
+    clean_zk_metadata_for_hosts(ctx, fqdn)
diff --git a/ch_tools/chadmin/internal/zookeeper.py b/ch_tools/chadmin/internal/zookeeper.py
index e5565395..dd4115b9 100644
--- a/ch_tools/chadmin/internal/zookeeper.py
+++ b/ch_tools/chadmin/internal/zookeeper.py
@@ -1,9 +1,11 @@
 import logging
 import os
+import re
 from contextlib import contextmanager
+from queue import Queue
 
 from kazoo.client import KazooClient
-from kazoo.exceptions import NoNodeError
+from kazoo.exceptions import NoNodeError, NotEmptyError
 
 from ch_tools.chadmin.cli import get_config, get_macros
 
@@ -27,19 +29,22 @@ def get_zk_node_acls(ctx, path):
         return zk.get_acls(path)
 
 
+def _get_children(zk, path):
+    try:
+        return zk.get_children(path)
+    except NoNodeError:
+        return []  # in the case ZK deletes a znode while we traverse the tree
+
+
 def list_zk_nodes(ctx, path, verbose=False):
     def _stat_node(zk, node):
         descendants_count = 0
         queue = [node]
         while queue:
             item = queue.pop()
-            try:
-                children = zk.get_children(item)
-                descendants_count += len(children)
-                queue.extend(os.path.join(item, node) for node in children)
-            except NoNodeError:
-                # ZooKeeper nodes can be deleted during node tree traversal
-                pass
+            children = _get_children(zk, item)
+            descendants_count += len(children)
+            queue.extend(os.path.join(item, node) for node in children)
 
         return {
             "path": node,
@@ -83,11 +88,15 @@ def delete_zk_node(ctx, path):
 
 
 def delete_zk_nodes(ctx, paths):
+    paths_formated = [_format_path(ctx, path) for path in paths]
     with zk_client(ctx) as zk:
-        for path in paths:
-            path = _format_path(ctx, path)
-            print(f"Deleting ZooKeeper node {path}")
-            zk.delete(path, recursive=True)
+        _delete_zk_nodes(zk, paths_formated)
+
+
+def _delete_zk_nodes(zk, paths):
+    for path in paths:
+        print(f"Deleting ZooKeeper node {path}")
+        zk.delete(path, recursive=True)
 
 
 def _format_path(ctx, path):
@@ -98,6 +107,136 @@ def _format_path(ctx, path):
     return path.format_map(get_macros(ctx))
 
 
+def _set_node_value(zk, path, value):
+    """
+    Set value to node in zk.
+    """
+    if zk.exists(path):
+        try:
+            zk.set(path, value.encode())
+        except NoNodeError:
+            print(f"Can not set for node: {path}  value : {value}")
+
+
+def _find_paths(zk, root_path, included_paths_regexp, excluded_paths_regexp=None):
+    """
+    Traverse zookeeper tree from root_path with bfs approach.
+
+    Return paths of nodes that match the include regular expression and do not match the excluded one.
+    """
+    paths = set()
+    queue: Queue = Queue()
+    queue.put(root_path)
+    included_regexp = re.compile("|".join(included_paths_regexp))
+    excluded_regexp = (
+        re.compile("|".join(excluded_paths_regexp)) if excluded_paths_regexp else None
+    )
+
+    while not queue.empty():
+        path = queue.get()
+        if excluded_regexp and re.match(excluded_regexp, path):
+            continue
+
+        for child_node in _get_children(zk, path):
+            subpath = os.path.join(path, child_node)
+
+            if re.match(included_regexp, subpath):
+                paths.add(subpath)
+            else:
+                queue.put(os.path.join(path, subpath))
+
+    return list(paths)
+
+
+def clean_zk_metadata_for_hosts(ctx, nodes):
+    """
+    Perform cleanup in zookeeper after deleting hosts in the cluster or whole cluster deleting.
+    """
+
+    def _try_delete_zk_node(zk, node):
+        try:
+            _delete_zk_nodes(zk, [node])
+        except NoNodeError:
+            #  Someone deleted node before us. Do nothing.
+            print("Node {node} is already absent, skipped".format(node=node))
+        except NotEmptyError:
+            #  Someone created child node while deleting.
+            #  I'm not sure that we can get this exception with recursive=True.
+            #  Do nothing.
+            print("Node {node} is not empty, skipped".format(node=node))
+
+    def _find_parents(zk, root_path, nodes, parent_name):
+        excluded_paths = [
+            ".*clickhouse/task_queue",
+            ".*clickhouse/zero_copy",
+        ]
+        included_paths = [".*/" + parent_name + "/" + node for node in nodes]
+        paths = _find_paths(zk, root_path, included_paths, excluded_paths)
+        # Paths will be like */shard1/replicas/hostname. But we need */shard1.
+        # Go up for 2 directories.
+        paths = [os.sep.join(path.split(os.sep)[:-2]) for path in paths]
+        # One path might be in list several times. Make list unique.
+        return list(set(paths))
+
+    def _set_replicas_is_lost(zk, table_paths, nodes):
+        """
+        Set flag <path>/replicas/<replica_name>/is_lost to 1
+        """
+        for path in table_paths:
+
+            replica_path = os.path.join(path, "replicas")
+            if not zk.exists(replica_path):
+                continue
+
+            for node in nodes:
+                is_lost_flag_path = os.path.join(replica_path, node, "is_lost")
+                print("Set is_lost_flag " + is_lost_flag_path)
+                _set_node_value(zk, is_lost_flag_path, "1")
+
+    def _remove_replicas_queues(zk, paths, replica_names):
+        """
+        Remove <path>/replicas/<replica_name>/queue
+        """
+        for path in paths:
+            replica_name = os.path.join(path, "replicas")
+            if not zk.exists(replica_name):
+                continue
+
+            for replica_name in replica_names:
+                queue_path = os.path.join(replica_name, replica_name, "queue")
+                if not zk.exists(queue_path):
+                    continue
+
+                if zk.exists(queue_path):
+                    _try_delete_zk_node(zk, queue_path)
+
+    def _nodes_absent(zk, zk_root_path, nodes):
+        included_paths = [".*/" + node for node in nodes]
+        paths_to_delete = _find_paths(zk, zk_root_path, included_paths)
+        for node in paths_to_delete:
+            _try_delete_zk_node(zk, node)
+
+    def _absent_if_empty_child(zk, path, child):
+        """
+        Remove node if subnode is empty
+        """
+        child_full_path = os.path.join(path, child)
+        if (
+            not zk.exists(child_full_path)
+            or len(_get_children(zk, child_full_path)) == 0
+        ):
+            _try_delete_zk_node(zk, path)
+
+    zk_root_path = _format_path(ctx, "/")
+    with zk_client(ctx) as zk:
+        table_paths = _find_parents(zk, zk_root_path, nodes, "replicas")
+        _set_replicas_is_lost(zk, table_paths, nodes)
+        _remove_replicas_queues(zk, table_paths, nodes)
+        _nodes_absent(zk, zk_root_path, nodes)
+        for path in table_paths:
+            _absent_if_empty_child(zk, path, "replicas")
+
+
 @contextmanager
 def zk_client(ctx):
     zk = _get_zk_client(ctx)
@@ -119,6 +258,7 @@ def _get_zk_client(ctx):
     zkcli_identity = args.get("zkcli_identity")
     no_chroot = args.get("no_chroot", False)
     no_ch_config = args.get("no_ch_config", False)
+    zk_root_path = args.get("zk_root_path", None)
 
     if no_ch_config:
         if not host:
@@ -132,7 +272,9 @@ def _get_zk_client(ctx):
             f'{host if host else node["host"]}:{port if port else node["port"]}'
             for node in zk_config.nodes
         )
-        if not no_chroot and zk_config.root is not None:
+        if zk_root_path:
+            connect_str += zk_root_path
+        elif not no_chroot and zk_config.root is not None:
             connect_str += zk_config.root
 
         if zkcli_identity is None:
diff --git a/tests/features/chadmin_zookeeper.feature b/tests/features/chadmin_zookeeper.feature
new file mode 100644
index 00000000..aac8e72b
--- /dev/null
+++ b/tests/features/chadmin_zookeeper.feature
@@ -0,0 +1,43 @@
+Feature: chadmin zookeeper commands.
+
+  Background:
+    Given default configuration
+    And a working s3
+    And a working zookeeper
+    And a working clickhouse on clickhouse01
+
+
+  Scenario: Cleanup all hosts
+    When we execute chadmin create zk nodes on zookeeper01
+    """
+    /test/write_sli_part/shard1/replicas/host1.net
+    /test/write_sli_part/shard1/replicas/host2.net
+    /test/read_sli_part/shard1/replicas/host1.net
+    /test/read_sli_part/shard1/replicas/host2.net
+    /test/write_sli_part/shard1/log
+    """
+    And we do hosts cleanup on zookeeper01 with fqdn host1.net,host2.net and zk root /test
+
+    Then the list of children on zookeeper01 for zk node /test/write_sli_part are empty
+    And the list of children on zookeeper01 for zk node /test/read_sli_part are empty
+
+  Scenario: Cleanup single host 
+    When we execute chadmin create zk nodes on zookeeper01
+    """
+    /test/write_sli_part/shard1/replicas/host1.net
+    /test/write_sli_part/shard1/replicas/host2.net
+    /test/read_sli_part/shard1/replicas/host1.net
+    /test/read_sli_part/shard1/replicas/host2.net
+    /test/write_sli_part/shard1/log
+    """
+    And we do hosts cleanup on zookeeper01 with fqdn host1.net and zk root /test
+    
+
+    Then the list of children on zookeeper01 for zk node /test/write_sli_part/shard1/replicas/ are equal to
+    """
+    /test/write_sli_part/shard1/replicas/host2.net
+    """
+    And the list of children on zookeeper01 for zk node  /test/read_sli_part/shard1/replicas/ are equal to
+    """
+    /test/read_sli_part/shard1/replicas/host2.net
+    """
diff --git a/tests/modules/chadmin.py b/tests/modules/chadmin.py
new file mode 100644
index 00000000..015bfa1f
--- /dev/null
+++ b/tests/modules/chadmin.py
@@ -0,0 +1,35 @@
+import logging
+
+
+class Chadmin:
+    def __init__(self, container):
+        self._container = container
+
+    def exec_cmd(self, cmd):
+        ch_admin_cmd = f"chadmin {cmd}"
+        logging.debug("chadmin command:", ch_admin_cmd)
+        result = self._container.exec_run(["bash", "-c", ch_admin_cmd], user="root")
+        return result
+
+    def create_zk_node(self, zk_node, no_ch_config=True, recursive=True):
+        cmd = "zookeeper {use_config} create {make_parents} {node}".format(
+            use_config="--no-ch-config" if no_ch_config else "",
+            make_parents="--make-parents" if recursive else "",
+            node=zk_node,
+        )
+        return self.exec_cmd(cmd)
+
+    def zk_list(self, zk_node, no_ch_config=True):
+        cmd = "zookeeper {use_config} list {node}".format(
+            use_config="--no-ch-config" if no_ch_config else "",
+            node=zk_node,
+        )
+        return self.exec_cmd(cmd)
+
+    def zk_cleanup(self, fqdn, zk_root, no_ch_config=True):
+        cmd = "zookeeper {use_config} --chroot {root} cleanup-removed-hosts-metadata {hosts}".format(
+            use_config="--no-ch-config" if no_ch_config else "",
+            root=zk_root,
+            hosts=fqdn,
+        )
+        return self.exec_cmd(cmd)
diff --git a/tests/steps/chadmin.py b/tests/steps/chadmin.py
new file mode 100644
index 00000000..072d165a
--- /dev/null
+++ b/tests/steps/chadmin.py
@@ -0,0 +1,40 @@
+"""
+Steps for interacting with chadmin.
+"""
+
+from behave import then, when
+from hamcrest import assert_that, equal_to
+from modules.chadmin import Chadmin
+from modules.docker import get_container
+
+
+@when("we execute chadmin create zk nodes on {node:w}")
+def step_create_(context, node):
+    container = get_container(context, node)
+    nodes = context.text.strip().split("\n")
+    chadmin = Chadmin(container)
+
+    for node in nodes:
+        result = chadmin.create_zk_node(node)
+        assert result.exit_code == 0, f" output:\n {result.output.decode().strip()}"
+
+
+@when("we do hosts cleanup on {node} with fqdn {fqdn} and zk root {zk_root}")
+def step_host_cleanup(context, node, fqdn, zk_root):
+    container = get_container(context, node)
+    result = Chadmin(container).zk_cleanup(fqdn, zk_root)
+    assert result.exit_code == 0, f" output:\n {result.output.decode().strip()}"
+
+
+@then("the list of children on {node:w} for zk node {zk_node} are equal to")
+def step_childen_list(context, node, zk_node):
+    container = get_container(context, node)
+    result = Chadmin(container).zk_list(zk_node)
+    assert_that(result.output.decode(), equal_to(context.text + "\n"))
+
+
+@then("the list of children on {node:w} for zk node {zk_node} are empty")
+def step_childen_list_empty(context, node, zk_node):
+    container = get_container(context, node)
+    result = Chadmin(container).zk_list(zk_node)
+    assert_that(result.output.decode(), equal_to("\n"))