Wait replication sync command

ch_tools/chadmin/chadmin_cli.py

@@ -42,7 +42,7 @@
 from ch_tools.chadmin.cli.table_group import table_group
 from ch_tools.chadmin.cli.table_replica_group import table_replica_group
 from ch_tools.chadmin.cli.thread_log_group import thread_log_group
-from ch_tools.chadmin.cli.wait_started_command import wait_started_command
+from ch_tools.chadmin.cli.wait_group import wait_group
 from ch_tools.chadmin.cli.zookeeper_group import zookeeper_group
 from ch_tools.common.cli.context_settings import CONTEXT_SETTINGS
 from ch_tools.common.cli.locale_resolver import LocaleResolver
@@ -116,7 +116,6 @@ def cli(ctx, format_, settings, timeout, port, debug):
     list_settings_command,
     restore_replica_command,
     stack_trace_command,
-    wait_started_command,
 ]
 
 groups: List[Any] = [
@@ -139,6 +138,7 @@ def cli(ctx, format_, settings, timeout, port, debug):
     table_group,
     table_replica_group,
     thread_log_group,
+    wait_group,
     zookeeper_group,
 ]
 

ch_tools/chadmin/cli/wait_started_command.py → ch_tools/chadmin/cli/wait_group.py

@@ -3,17 +3,106 @@
 import sys
 import time
 
-from click import command, option, pass_context
+from click import FloatRange, group, option, pass_context
 
 from ch_tools.chadmin.internal.utils import execute_query
+from ch_tools.common.cli.parameters import TimeSpanParamType
+from ch_tools.common.commands.replication_lag import estimate_replication_lag
 from ch_tools.common.utils import execute
 
 BASE_TIMEOUT = 600
 LOCAL_PART_LOAD_SPEED = 10  # in data parts per second
 S3_PART_LOAD_SPEED = 0.5  # in data parts per second
 
 
-@command("wait-started")
+@group("wait")
+def wait_group():
+    """Commands to wait until Clickhouse is in a certain state."""
+    pass
+
+
+@wait_group.command("replication-sync")
+@option(
+    "-s",
+    "--status",
+    type=int,
+    default=0,
+    help="Wait until replication-lag returned status is no worse than given, 0 = OK, 1 = WARN, 2 = CRIT.",
+)
+@option(
+    "-p",
+    "--pause",
+    type=TimeSpanParamType(),
+    default="30s",
+    help="Pause between requests.",
+)
+@option(
+    "-t",
+    "--timeout",
+    type=TimeSpanParamType(),
+    default="3d",
+    help="Max amount of time to wait.",
+)
+@option(
+    "-x",
+    "--exec-critical",
+    "xcrit",
+    type=int,
+    default=3600,
+    help="Critical threshold for one task execution.",
+)
+@option(
+    "-c",
+    "--critical",
+    "crit",
+    type=int,
+    default=600,
+    help="Critical threshold for lag with errors.",
+)
+@option("-w", "--warning", "warn", type=int, default=300, help="Warning threshold.")
+@option(
+    "-M",
+    "--merges-critical",
+    "mcrit",
+    type=FloatRange(0.0, 100.0),
+    default=90.0,
+    help="Critical threshold in percent of max_replicated_merges_in_queue.",
+)
+@option(
+    "-m",
+    "--merges-warning",
+    "mwarn",
+    type=FloatRange(0.0, 100.0),
+    default=50.0,
+    help="Warning threshold in percent of max_replicated_merges_in_queue.",
+)
+@option(
+    "-v",
+    "--verbose",
+    "verbose",
+    type=int,
+    count=True,
+    default=0,
+    help="Show details about lag.",
+)
+@pass_context
+def wait_replication_sync_command(
+    ctx, status, pause, timeout, xcrit, crit, warn, mwarn, mcrit, verbose
+):
+    """Wait for ClickHouse server to sync replication with other replicas using replication-lag command."""
+
+    deadline = time.time() + timeout.total_seconds()
+    while time.time() < deadline:
+        res = estimate_replication_lag(ctx, xcrit, crit, warn, mwarn, mcrit, verbose)
+        if res.code <= status:
+            sys.exit(0)
+        time.sleep(pause.total_seconds())
+
+    logging.error("ClickHouse can't sync replicas.")
+    sys.exit(1)
+
+
+@wait_group.command("started")
 @option(
     "--timeout",
     type=int,