Merge pull request #8 from exasol/multinode

Multinode

Author: ooke

benchkit/cli.py

@@ -282,6 +282,244 @@ def status(
         _show_configs_summary(config_files)
 
 
+@app.command()
+def check(
+    config: str = typer.Option(..., "--config", "-c", help="Path to config YAML file"),
+    verbose: bool = typer.Option(
+        False, "--verbose", "-v", help="Show all configuration details"
+    ),
+) -> None:
+    """Check and display configuration file contents."""
+    from pathlib import Path
+
+    import yaml
+    from rich.panel import Panel
+    from rich.text import Text
+
+    config_path = Path(config)
+
+    # Try to load and validate the config
+    validation_errors: list[str] = []
+    cfg: dict[str, Any] | None = None
+
+    try:
+        cfg = load_config(config)
+    except FileNotFoundError as e:
+        console.print(
+            Panel(
+                f"[red]Configuration file not found:[/red] {config_path}",
+                title="Configuration Check",
+                border_style="red",
+            )
+        )
+        raise typer.Exit(1) from e
+    except ValueError as e:
+        # Parse validation errors from the exception
+        error_msg = str(e)
+        validation_errors.append(error_msg)
+    except Exception as e:
+        validation_errors.append(f"Unexpected error: {e}")
+
+    # Display header with validation status
+    if validation_errors:
+        status_text = Text()
+        status_text.append("Configuration: ", style="bold")
+        status_text.append(str(config_path), style="cyan")
+        status_text.append("\nStatus: ", style="bold")
+        status_text.append("✗ Invalid", style="red bold")
+        console.print(Panel(status_text, border_style="red"))
+
+        console.print("\n[red bold]Errors found:[/red bold]")
+        for i, error in enumerate(validation_errors, 1):
+            # Clean up the error message
+            error = error.replace("Invalid configuration: ", "")
+            console.print(f"  {i}. {error}")
+
+        raise typer.Exit(1)
+
+    # Config is valid - display it
+    status_text = Text()
+    status_text.append("Configuration: ", style="bold")
+    status_text.append(str(config_path), style="cyan")
+    status_text.append("\nStatus: ", style="bold")
+    status_text.append("✓ Valid", style="green bold")
+    console.print(Panel(status_text, border_style="green"))
+
+    if cfg is None:
+        raise typer.Exit(1)
+
+    # Display Project section
+    project_table = Table(show_header=False, box=None, padding=(0, 2))
+    project_table.add_column("Key", style="bold")
+    project_table.add_column("Value")
+    project_table.add_row("ID", cfg.get("project_id", "-"))
+    project_table.add_row("Title", cfg.get("title", "-"))
+    project_table.add_row("Author", cfg.get("author", "-"))
+    console.print(Panel(project_table, title="Project", border_style="blue"))
+
+    # Display Environment section
+    env = cfg.get("env", {})
+    env_table = Table(show_header=False, box=None, padding=(0, 2))
+    env_table.add_column("Key", style="bold")
+    env_table.add_column("Value")
+    env_table.add_row("Mode", env.get("mode", "local"))
+    if env.get("region"):
+        env_table.add_row("Region", env.get("region"))
+    env_table.add_row(
+        "External DB Access", "Yes" if env.get("allow_external_database_access") else "No"
+    )
+    if env.get("ssh_key_name"):
+        env_table.add_row("SSH Key", env.get("ssh_key_name"))
+    console.print(Panel(env_table, title="Environment", border_style="blue"))
+
+    # Display Systems section
+    systems = cfg.get("systems", [])
+    instances = env.get("instances", {}) or {}
+
+    # Collect all system names for IP variable validation display
+    system_names_upper = {s["name"].upper() for s in systems}
+
+    systems_content = []
+    for i, system in enumerate(systems, 1):
+        name = system.get("name", "unnamed")
+        kind = system.get("kind", "unknown")
+        version = system.get("version", "unknown")
+        setup = system.get("setup", {})
+        method = setup.get("method", "default")
+        node_count = setup.get("node_count", 1)
+
+        system_lines = [
+            f"[bold cyan]{i}. {name}[/bold cyan]",
+            f"   Kind:     {kind}",
+            f"   Version:  {version}",
+            f"   Method:   {method}",
+        ]
+
+        # Node count
+        node_desc = "single-node" if node_count == 1 else f"{node_count}-node cluster"
+        system_lines.append(f"   Nodes:    {node_count} ({node_desc})")
+
+        # Instance type if available
+        instance_config = instances.get(name, {})
+        if instance_config.get("instance_type"):
+            system_lines.append(f"   Instance: {instance_config.get('instance_type')}")
+
+        # IP variables
+        ip_vars = []
+        ip_fields = ["host", "host_addrs", "host_external_addrs"]
+        for field in ip_fields:
+            value = setup.get(field, "")
+            if isinstance(value, str) and value.startswith("$"):
+                ip_vars.append(value)
+
+        if ip_vars:
+            # Check if IP vars are valid
+            all_valid = True
+            for var in ip_vars:
+                var_name = var[1:]  # Remove $
+                import re
+
+                match = re.match(r"^([A-Z_][A-Z0-9_]*)_(PRIVATE|PUBLIC)_IP$", var_name)
+                if match and match.group(1) not in system_names_upper:
+                    all_valid = False
+                    break
+
+            status = "[green]✓[/green]" if all_valid else "[red]✗[/red]"
+            system_lines.append(f"   IP Vars:  {', '.join(ip_vars)} {status}")
+
+        # Verbose mode: show all setup params
+        if verbose:
+            system_lines.append("   [dim]Setup:[/dim]")
+            for key, value in setup.items():
+                if key not in ["method", "node_count"]:
+                    # Mask passwords
+                    if "password" in key.lower():
+                        value = "********"
+                    system_lines.append(f"      {key}: {value}")
+
+        systems_content.append("\n".join(system_lines))
+
+    console.print(
+        Panel(
+            "\n\n".join(systems_content),
+            title=f"Systems ({len(systems)})",
+            border_style="blue",
+        )
+    )
+
+    # Display Workload section
+    workload = cfg.get("workload", {})
+    workload_table = Table(show_header=False, box=None, padding=(0, 2))
+    workload_table.add_column("Key", style="bold")
+    workload_table.add_column("Value")
+    workload_table.add_row("Name", workload.get("name", "-"))
+    sf = workload.get("scale_factor", 1)
+    workload_table.add_row("Scale Factor", f"{sf} ({sf} GB for TPC-H)")
+    workload_table.add_row("Data Format", workload.get("data_format", "csv"))
+    workload_table.add_row("Runs/Query", str(workload.get("runs_per_query", 3)))
+    workload_table.add_row("Warmup Runs", str(workload.get("warmup_runs", 1)))
+
+    # Query info
+    queries = workload.get("queries", {})
+    include = queries.get("include", [])
+    exclude = queries.get("exclude", [])
+    if include:
+        workload_table.add_row("Queries", f"Include: {', '.join(include)}")
+    elif exclude:
+        workload_table.add_row("Queries", f"All except: {', '.join(exclude)}")
+    else:
+        workload_table.add_row("Queries", "All (22 queries for TPC-H)")
+
+    # Multiuser
+    multiuser = workload.get("multiuser")
+    if multiuser and multiuser.get("enabled", True):
+        num_streams = multiuser.get("num_streams", 1)
+        randomize = multiuser.get("randomize", False)
+        random_seed = multiuser.get("random_seed", "-")
+        workload_table.add_row(
+            "Multiuser",
+            f"{num_streams} streams, randomize: {randomize}, seed: {random_seed}",
+        )
+
+    console.print(Panel(workload_table, title="Workload", border_style="blue"))
+
+    # Display Execution section
+    execution = cfg.get("execution", {})
+    if execution.get("parallel"):
+        exec_table = Table(show_header=False, box=None, padding=(0, 2))
+        exec_table.add_column("Key", style="bold")
+        exec_table.add_column("Value")
+        exec_table.add_row("Parallel", "Yes")
+        if execution.get("max_workers"):
+            exec_table.add_row("Max Workers", str(execution.get("max_workers")))
+        console.print(Panel(exec_table, title="Execution", border_style="blue"))
+
+    # Display Report section
+    report = cfg.get("report", {})
+    if report:
+        report_table = Table(show_header=False, box=None, padding=(0, 2))
+        report_table.add_column("Key", style="bold")
+        report_table.add_column("Value")
+        report_table.add_row("Output", report.get("output_path", "-"))
+        report_table.add_row("Figures", report.get("figures_dir", "-"))
+
+        # Charts enabled
+        charts = []
+        if report.get("show_boxplots", True):
+            charts.append("boxplots")
+        if report.get("show_latency_cdf"):
+            charts.append("latency CDF")
+        if report.get("show_bar_chart", True):
+            charts.append("bar chart")
+        if report.get("show_heatmap", True):
+            charts.append("heatmap")
+        report_table.add_row("Charts", ", ".join(charts) if charts else "none")
+
+        console.print(Panel(report_table, title="Report Settings", border_style="blue"))
+
+    console.print("\n[green]Configuration is valid and ready to use.[/green]")
+
+
 @app.command()
 def infra(
     action: str = typer.Argument(..., help="Action: plan, apply, destroy"),
@@ -819,21 +1057,38 @@ def _probe_remote_systems(config: dict[str, Any], outdir: Path) -> bool:
         outputs = result.outputs or {}
 
         # Extract instance information from new terraform output format
-        instances = {}
+        # After Terraform fix, IPs are always lists: ["ip"] for single-node, ["ip1", "ip2"] for multinode
+        instances_to_probe = (
+            []
+        )  # List of (system_name, node_idx, public_ip, private_ip)
 
-        # New format: system_public_ips = {"exasol": "ip", "clickhouse": "ip"}
+        # New format: system_public_ips = {"exasol": ["ip"], "clickhouse": ["ip1", "ip2"]}
         # Note: _parse_terraform_outputs already extracted the "value" field
         if "system_public_ips" in outputs:
             public_ips = outputs["system_public_ips"] or {}
             private_ips = outputs.get("system_private_ips", {}) or {}
 
-            for system_name, public_ip in public_ips.items():
-                instances[system_name] = {
-                    "public_ip": public_ip,
-                    "private_ip": private_ips.get(system_name),
-                }
+            for system_name, public_ip_list in public_ips.items():
+                private_ip_list = private_ips.get(system_name)
+
+                # Handle both list and single IP (backward compatibility)
+                if isinstance(public_ip_list, list):
+                    for idx, public_ip in enumerate(public_ip_list):
+                        private_ip = (
+                            private_ip_list[idx]
+                            if isinstance(private_ip_list, list)
+                            else private_ip_list
+                        )
+                        instances_to_probe.append(
+                            (system_name, idx, public_ip, private_ip)
+                        )
+                else:
+                    # Backward compatibility: single IP (not a list)
+                    instances_to_probe.append(
+                        (system_name, 0, public_ip_list, private_ip_list)
+                    )
 
-        if not instances:
+        if not instances_to_probe:
             console.print("[yellow]No instances found in terraform outputs[/yellow]")
             return False
 
@@ -848,19 +1103,31 @@ def _probe_remote_systems(config: dict[str, Any], outdir: Path) -> bool:
         ssh_key_path = os.path.expanduser(ssh_key_path)
 
         success_count = 0
-        total_instances = len(instances)
-
-        for system_name, instance_info in instances.items():
-            public_ip = instance_info["public_ip"]
-            console.print(f"[blue]Probing {system_name} ({public_ip})...[/blue]")
+        total_instances = len(instances_to_probe)
+
+        for system_name, node_idx, public_ip, private_ip in instances_to_probe:
+            # Show node index for multinode systems
+            node_label = (
+                f"-node{node_idx}"
+                if any(
+                    s == system_name and i != node_idx
+                    for s, i, _, _ in instances_to_probe
+                )
+                else ""
+            )
+            console.print(
+                f"[blue]Probing {system_name}{node_label} ([{public_ip}])...[/blue]"
+            )
 
             if _probe_single_remote_system(
-                system_name, public_ip, ssh_key_path, ssh_user, outdir
+                f"{system_name}{node_label}", public_ip, ssh_key_path, ssh_user, outdir
             ):
-                console.print(f"[green]✓ {system_name} probe completed[/green]")
+                console.print(
+                    f"[green]✓ {system_name}{node_label} probe completed[/green]"
+                )
                 success_count += 1
             else:
-                console.print(f"[red]✗ {system_name} probe failed[/red]")
+                console.print(f"[red]✗ {system_name}{node_label} probe failed[/red]")
 
         console.print(
             f"[blue]Completed {success_count}/{total_instances} system probes[/blue]"