Admin/XMover: Add problematic translog detection and cancellation

The new command can find and cancel replica shards with large
uncommitted translog sizes. It provides options to set size thresholds
and optionally execute cancel commands after user confirmation.

Author: amotl

cratedb_toolkit/admin/xmover/analysis/shard.py

@@ -838,6 +838,106 @@ def plan_node_decommission(self, node_name: str, min_free_space_gb: float = 100.
         }
 
 
+class TranslogReporter:
+    def __init__(self, client: CrateDBClient):
+        self.client = client
+
+    def problematic_translogs(self, size_mb: int) -> List[str]:
+        """Find and optionally cancel shards with problematic translog sizes."""
+        console.print(Panel.fit("[bold blue]Problematic Translog Analysis[/bold blue]"))
+        console.print(f"[dim]Looking for replica shards with translog uncommitted size > {size_mb}MB[/dim]")
+        console.print()
+
+        # Query to find problematic replica shards
+        query = """
+                SELECT sh.schema_name, \
+                       sh.table_name, \
+                       translate(p.values::text, ':{}', '=()') as partition_values, \
+                       sh.id                                   AS shard_id, \
+                       node['name']                            as node_name, \
+                       sh.translog_stats['uncommitted_size'] / 1024^2 AS translog_uncommitted_mb
+                FROM
+                    sys.shards AS sh
+                    LEFT JOIN information_schema.table_partitions p
+                ON sh.table_name = p.table_name
+                    AND sh.schema_name = p.table_schema
+                    AND sh.partition_ident = p.partition_ident
+                WHERE
+                    sh.state = 'STARTED'
+                  AND sh.translog_stats['uncommitted_size'] \
+                    > ? * 1024^2
+                  AND primary = FALSE
+                ORDER BY
+                    6 DESC \
+                """
+
+        try:
+            result = self.client.execute_query(query, [size_mb])
+            rows = result.get("rows", [])
+
+            if not rows:
+                console.print(f"[green]✓ No replica shards found with translog uncommitted size > {size_mb}MB[/green]")
+                return []
+
+            console.print(f"Found {len(rows)} shards with problematic translogs:")
+            console.print()
+
+            # Display query results table
+            results_table = Table(title=f"Problematic Replica Shards (translog > {size_mb}MB)", box=box.ROUNDED)
+            results_table.add_column("Schema", style="cyan")
+            results_table.add_column("Table", style="blue")
+            results_table.add_column("Partition", style="magenta")
+            results_table.add_column("Shard ID", justify="right", style="yellow")
+            results_table.add_column("Node", style="green")
+            results_table.add_column("Translog MB", justify="right", style="red")
+
+            for row in rows:
+                schema_name, table_name, partition_values, shard_id, node_name, translog_mb = row
+                partition_display = (
+                    partition_values if partition_values and partition_values != "NULL" else "[dim]none[/dim]"
+                )
+                results_table.add_row(
+                    schema_name, table_name, partition_display, str(shard_id), node_name, f"{translog_mb:.1f}"
+                )
+
+            console.print(results_table)
+            console.print()
+            console.print("[bold]Generated ALTER Commands:[/bold]")
+            console.print()
+
+            # Generate ALTER commands
+            alter_commands = []
+            for row in rows:
+                schema_name, table_name, partition_values, shard_id, node_name, translog_mb = row
+
+                # Build the ALTER command based on whether it's partitioned
+                if partition_values and partition_values != "NULL":
+                    # partition_values already formatted like ("sync_day"=1757376000000) from the translate function
+                    alter_cmd = (
+                        f'ALTER TABLE "{schema_name}"."{table_name}" partition {partition_values} '
+                        f"REROUTE CANCEL SHARD {shard_id} on '{node_name}' WITH (allow_primary=False);"
+                    )
+                else:
+                    alter_cmd = (
+                        f'ALTER TABLE "{schema_name}"."{table_name}" '
+                        f"REROUTE CANCEL SHARD {shard_id} on '{node_name}' WITH (allow_primary=False);"
+                    )
+
+                alter_commands.append(alter_cmd)
+                console.print(alter_cmd)
+
+            console.print()
+            console.print(f"[bold]Total: {len(alter_commands)} ALTER commands generated[/bold]")
+            return alter_commands
+
+        except Exception as e:
+            console.print(f"[red]Error analyzing problematic translogs: {e}[/red]")
+            import traceback
+
+            console.print(f"[dim]{traceback.format_exc()}[/dim]")
+            return []
+
+
 class ShardReporter:
     def __init__(self, analyzer: ShardAnalyzer):
         self.analyzer = analyzer

cratedb_toolkit/admin/xmover/cli.py

@@ -12,7 +12,12 @@
 from rich.console import Console
 from rich.panel import Panel
 
-from cratedb_toolkit.admin.xmover.analysis.shard import ActiveShardMonitor, ShardAnalyzer, ShardReporter
+from cratedb_toolkit.admin.xmover.analysis.shard import (
+    ActiveShardMonitor,
+    ShardAnalyzer,
+    ShardReporter,
+    TranslogReporter,
+)
 from cratedb_toolkit.admin.xmover.analysis.table import DistributionAnalyzer
 from cratedb_toolkit.admin.xmover.analysis.zone import ZoneReport
 from cratedb_toolkit.admin.xmover.model import (
@@ -444,6 +449,69 @@ def run_single_analysis():
         console.print(f"[dim]{traceback.format_exc()}[/dim]")
 
 
+@main.command()
+@click.option("--size-mb", default=300, help="Minimum translog uncommitted size in MB (default: 300)")
+@click.option("--cancel", is_flag=True, help="Execute the cancel commands after confirmation")
+@click.pass_context
+def problematic_translogs(ctx, size_mb: int, cancel: bool):
+    """
+    Find and optionally cancel shards with problematic translog sizes.
+
+    This command identifies replica shards with large uncommitted translog sizes
+    that may indicate replication issues. By default, it shows the ALTER commands
+    that would cancel these shards. With --cancel, it executes them after confirmation.
+    """
+    client = ctx.obj["client"]
+    report = TranslogReporter(client=client)
+    alter_commands = report.problematic_translogs(size_mb=size_mb)
+
+    try:
+        if cancel and alter_commands:
+            console.print()
+            console.print("[yellow]⚠️  WARNING: This will cancel the specified shards![/yellow]")
+            console.print("[yellow]This may cause temporary data unavailability for these shards.[/yellow]")
+            console.print()
+
+            if click.confirm("Are you sure you want to execute these ALTER commands?"):
+                console.print()
+                console.print("[bold blue]Executing ALTER commands...[/bold blue]")
+
+                executed = 0
+                failed = 0
+
+                for i, cmd in enumerate(alter_commands, 1):
+                    if cmd.startswith("--"):
+                        console.print(f"[yellow]Skipping command {i} (parse error): {cmd}[/yellow]")
+                        continue
+
+                    try:
+                        console.print(f"[dim]({i}/{len(alter_commands)}) Executing...[/dim]")
+                        client.execute_query(cmd)
+                        console.print(f"[green]✓ Command {i} executed successfully[/green]")
+                        executed += 1
+                    except Exception as e:
+                        console.print(f"[red]✗ Command {i} failed: {e}[/red]")
+                        failed += 1
+
+                    # Small delay between commands to avoid overwhelming the cluster
+                    if i < len(alter_commands):
+                        time.sleep(1)
+
+                console.print()
+                console.print("[bold]Execution Summary:[/bold]")
+                console.print(f"[green]✓ Successful: {executed}[/green]")
+                if failed > 0:
+                    console.print(f"[red]✗ Failed: {failed}[/red]")
+            else:
+                console.print("[yellow]Operation cancelled by user[/yellow]")
+
+    except Exception as e:
+        console.print(f"[red]Error analyzing problematic translogs: {e}[/red]")
+        import traceback
+
+        console.print(f"[dim]{traceback.format_exc()}[/dim]")
+
+
 @main.command()
 @click.option("--table", "-t", help="Analyze zones for specific table only")
 @click.option("--show-shards/--no-show-shards", default=False, help="Show individual shard details (default: False)")

doc/admin/xmover/handbook.md

@@ -245,6 +245,47 @@ xmover monitor-recovery --watch --include-transitioning
 - **DISK**: Rebuilding shard from local data (after restart/disk issues)
 
 
+### `problematic-translogs`
+Find and optionally cancel replica shards with problematic translog sizes.
+
+**Options:**
+- `--size-mb INTEGER`: Minimum translog uncommitted size in MB (default: 300)
+- `--cancel`: Execute the cancel commands after confirmation
+
+**Description:**
+This command identifies replica shards with large uncommitted translog sizes that may indicate replication issues. By default, it shows the ALTER commands that would cancel these shards. With `--cancel`, it executes them after confirmation.
+
+**Examples:**
+```bash
+# Show problematic shards with translog > 300MB (default)
+xmover problematic-translogs
+
+# Show shards with translog > 500MB
+xmover problematic-translogs --size-mb 500
+
+# Execute cancel commands for shards > 1GB after confirmation
+xmover problematic-translogs --size-mb 1000 --cancel
+```
+
+**Sample Output:**
+```
+Found 3 shards with problematic translogs:
+                   Problematic Replica Shards (translog > 300MB)
+╭────────┬───────────────────────────────┬────────────────────────────┬──────────┬────────────┬─────────────╮
+│ Schema │ Table                         │ Partition                  │ Shard ID │ Node       │ Translog MB │
+├────────┼───────────────────────────────┼────────────────────────────┼──────────┼────────────┼─────────────┤
+│ TURVO  │ shipmentFormFieldData         │ none                       │       14 │ data-hot-6 │      7040.9 │
+│ TURVO  │ shipmentFormFieldData_events  │ ("sync_day"=1757376000000) │        3 │ data-hot-2 │       481.2 │
+│ TURVO  │ orderFormFieldData            │ none                       │        5 │ data-hot-1 │       469.5 │
+╰────────┴───────────────────────────────┴────────────────────────────┴──────────┴────────────┴─────────────╯
+Generated ALTER Commands:
+ALTER TABLE "TURVO"."shipmentFormFieldData" REROUTE CANCEL SHARD 14 on 'data-hot-6' WITH (allow_primary=False);
+ALTER TABLE "TURVO"."shipmentFormFieldData_events" partition ("sync_day"=1757376000000) REROUTE CANCEL SHARD 3 on 'data-hot-2' WITH (allow_primary=False);
+ALTER TABLE "TURVO"."orderFormFieldData" REROUTE CANCEL SHARD 5 on 'data-hot-1' WITH (allow_primary=False);
+Total: 3 ALTER commands generated
+```
+
+
 ### `active-shards`
 Monitor the most active shards by tracking checkpoint progression over time.
 This command helps identify which shards are receiving the most write activity