diff --git a/README.md b/README.md index 1b364c9..c0611b7 100644 --- a/README.md +++ b/README.md @@ -429,7 +429,7 @@ clickhousectl cloud service update \ --transparent-data-encryption-key-id tde-key-1 \ --enable-core-dumps false -# Update replica scaling +# Update replica scaling (vertical autoscaling — fixed replica count, variable memory) clickhousectl cloud service scale \ --min-replica-memory-gb 24 \ --max-replica-memory-gb 48 \ @@ -437,6 +437,13 @@ clickhousectl cloud service scale \ --idle-scaling true \ --idle-timeout-minutes 10 +# Horizontal autoscaling — fixed memory per replica, variable replica count +# (requires the horizontal autoscaling org feature) +clickhousectl cloud service create --name my-service \ + --min-replicas 2 --max-replicas 8 --autoscaling-mode horizontal +clickhousectl cloud service scale \ + --min-replicas 2 --max-replicas 8 --autoscaling-mode horizontal + # Reset password with generated credentials clickhousectl cloud service reset-password @@ -480,9 +487,12 @@ clickhousectl cloud service delete --force | `--name` | Service name (required) | | `--provider` | Cloud provider: aws, gcp, azure (default: aws) | | `--region` | Region (default: us-east-1) | -| `--min-replica-memory-gb` | Min memory per replica in GB (8-356, multiple of 4) | -| `--max-replica-memory-gb` | Max memory per replica in GB (8-356, multiple of 4) | -| `--num-replicas` | Number of replicas (1-20) | +| `--min-replica-memory-gb` | Min memory per replica in GB (8-356, multiple of 4). Horizontal autoscaling requires it equal to `--max-replica-memory-gb` | +| `--max-replica-memory-gb` | Max memory per replica in GB (8-356, multiple of 4). Horizontal autoscaling requires it equal to `--min-replica-memory-gb` | +| `--num-replicas` | Number of replicas (1-20) (vertical autoscaling; mutually exclusive with `--min-replicas`/`--max-replicas`) | +| `--min-replicas` | Min number of replicas for horizontal autoscaling (mutually exclusive with `--num-replicas`) | +| `--max-replicas` | Max number of replicas for horizontal autoscaling (mutually exclusive with `--num-replicas`) | +| `--autoscaling-mode` | Autoscaling mode: `vertical` (default) or `horizontal`. Horizontal uses fixed memory per replica (`--min-replica-memory-gb` equal to `--max-replica-memory-gb`) with a variable replica count (`--min-replicas`/`--max-replicas`); vertical uses a fixed replica count (`--num-replicas`) with variable memory. On `service scale`, combine with the target mode's flags to switch modes in one call | | `--idle-scaling` | Allow scale to zero (default: true) | | `--idle-timeout-minutes` | Min idle timeout in minutes (>= 5) | | `--ip-allow` | IP CIDR to allow (repeatable, default: 0.0.0.0/0) | @@ -628,7 +638,7 @@ clickhousectl cloud clickpipe settings update \ Each source type has its own subcommand under `clickpipe create`: ```bash -# From S3 / object storage +# From S3 / object storage (one-shot snapshot) clickhousectl cloud clickpipe create object-storage \ --name my-s3-pipe \ --source-url 'https://bucket.s3.us-east-1.amazonaws.com/data/**' \ @@ -636,6 +646,19 @@ clickhousectl cloud clickpipe create object-storage \ --database default --table events \ --column "event_id:Int64" --column "name:String" +# From S3 with continuous ingestion (SQS queue) and ingestion control +# --skip-initial-load: skip the initial snapshot load, only ingest new objects +# --start-after: resume ingestion after a specific object key (conflicts with --skip-initial-load) +clickhousectl cloud clickpipe create object-storage \ + --name my-s3-continuous-pipe \ + --source-url 'https://bucket.s3.us-east-1.amazonaws.com/data/**' \ + --format JSONEachRow \ + --continuous \ + --queue-url 'https://sqs.us-east-1.amazonaws.com/123/my-queue' \ + --start-after obj-key-001 \ + --database default --table events \ + --column "event_id:Int64" --column "name:String" + # From Google Cloud Storage (object storage) clickhousectl cloud clickpipe create object-storage \ --name my-gcs-pipe \ @@ -675,11 +698,14 @@ clickhousectl cloud clickpipe create postgres \ --table-mapping "public.orders:public_orders" # From MySQL (CDC) +# --server-id sets the replication server ID (useful when multiple pipes read +# from the same MySQL instance, or to avoid colliding with existing replicas) clickhousectl cloud clickpipe create mysql \ --name my-mysql-pipe \ --host mysql.example.com \ --username root --password pass \ - --table-mapping "mydb.users:mydb_users" + --table-mapping "mydb.users:mydb_users" \ + --server-id 4242 # From MongoDB (CDC) clickhousectl cloud clickpipe create mongodb \ @@ -698,6 +724,28 @@ clickhousectl cloud clickpipe create bigquery \ Use `clickhousectl cloud clickpipe create --help` for the full list of options per source type. +#### Discovering a source schema (beta) + +`clickpipe schema-discover` probes a Kafka or Kinesis source and returns the +inferred fields/types without creating a pipe. It takes the same source +connection flags as the corresponding `create` subcommand (minus the +destination `--name`/`--database`/`--table`/`--column` options): + +```bash +# Discover schema from Kafka +clickhousectl cloud clickpipe schema-discover kafka \ + --brokers 'broker:9092' --topics events \ + --format JSONEachRow \ + --auth SCRAM-SHA-256 --username user --password pass + +# Discover schema from Kinesis +clickhousectl cloud clickpipe schema-discover kinesis \ + --stream-name events --region us-east-1 \ + --format JSONEachRow +``` + +Add `--json` (or run as a coding agent) for machine-readable output. + ### Members ```bash diff --git a/crates/clickhouse-cloud-api/src/models.rs b/crates/clickhouse-cloud-api/src/models.rs index 6327d82..9d6cacc 100644 --- a/crates/clickhouse-cloud-api/src/models.rs +++ b/crates/clickhouse-cloud-api/src/models.rs @@ -5204,6 +5204,11 @@ impl std::fmt::Display for AutoscalingMode { } } +impl AutoscalingMode { + /// Wire values accepted by the API, excluding the catch-all. + pub const VALUES: &'static [&'static str] = &["vertical", "horizontal"]; +} + /// Inline enum for `CurrentScaling.effectiveAutoscalingMode`. #[derive(Debug, Clone, PartialEq, Default, Serialize, Deserialize)] pub enum CurrentScalingEffectiveautoscalingmode { diff --git a/crates/clickhousectl/src/cloud/cli.rs b/crates/clickhousectl/src/cloud/cli.rs index 956c550..114eeac 100644 --- a/crates/clickhousectl/src/cloud/cli.rs +++ b/crates/clickhousectl/src/cloud/cli.rs @@ -355,6 +355,11 @@ impl CloudCommands { ClickPipeCommands::Stop { .. } => true, ClickPipeCommands::Resync { .. } => true, ClickPipeCommands::Scale { .. } => true, + // Side-effect-free, but the API gateway rejects OAuth/JWT on + // POST /clickpipes/schemaDiscovery ("This endpoint is not + // available for JWT authentication"), so classify it as a + // write to fail fast with the API-key guidance. + ClickPipeCommands::SchemaDiscover { .. } => true, ClickPipeCommands::Create { .. } => true, ClickPipeCommands::Settings { command } => match command { ClickPipeSettingsCommands::Get { .. } => false, @@ -492,18 +497,43 @@ CONTEXT FOR AGENTS: #[arg(long, default_value = "us-east-1")] region: String, - /// Minimum memory per replica in GB (8-356, multiple of 4) + /// Minimum memory per replica in GB (8-356, multiple of 4). Horizontal + /// autoscaling requires it equal to --max-replica-memory-gb. #[arg(long)] min_replica_memory_gb: Option, - /// Maximum memory per replica in GB (8-356, multiple of 4) + /// Maximum memory per replica in GB (8-356, multiple of 4). Horizontal + /// autoscaling requires it equal to --min-replica-memory-gb. #[arg(long)] max_replica_memory_gb: Option, - /// Number of replicas (1-20) - #[arg(long)] + /// Number of replicas (1-20). Vertical autoscaling; mutually exclusive + /// with the horizontal band (--min-replicas/--max-replicas). + #[arg(long, conflicts_with_all = ["min_replicas", "max_replicas"])] num_replicas: Option, + /// Minimum number of replicas for horizontal autoscaling (requires the + /// horizontal autoscaling org feature). Mutually exclusive with --num-replicas. + #[arg(long, conflicts_with = "num_replicas")] + min_replicas: Option, + + /// Maximum number of replicas for horizontal autoscaling (requires the + /// horizontal autoscaling org feature). Mutually exclusive with --num-replicas. + #[arg(long, conflicts_with = "num_replicas")] + max_replicas: Option, + + /// Autoscaling mode: vertical (default) or horizontal. Horizontal uses fixed + /// memory per replica (--min-replica-memory-gb equal to --max-replica-memory-gb) + /// with a variable replica count (--min-replicas/--max-replicas); vertical uses + /// a fixed replica count (--num-replicas) with variable memory. + #[arg( + long, + value_parser = PossibleValuesParser::new( + clickhouse_cloud_api::models::AutoscalingMode::VALUES + ) + )] + autoscaling_mode: Option, + /// Allow scale to zero when idle (default: true) #[arg(long)] idle_scaling: Option, @@ -691,18 +721,41 @@ CONTEXT FOR AGENTS: /// Service ID service_id: String, - /// Minimum memory per replica in GB (8-356, multiple of 4) + /// Minimum memory per replica in GB (8-356, multiple of 4). Horizontal + /// autoscaling requires it equal to --max-replica-memory-gb. #[arg(long)] min_replica_memory_gb: Option, - /// Maximum memory per replica in GB (8-356, multiple of 4) + /// Maximum memory per replica in GB (8-356, multiple of 4). Horizontal + /// autoscaling requires it equal to --min-replica-memory-gb. #[arg(long)] max_replica_memory_gb: Option, - /// Number of replicas (1-20) - #[arg(long)] + /// Number of replicas (1-20). Vertical autoscaling; mutually exclusive + /// with the horizontal band (--min-replicas/--max-replicas). + #[arg(long, conflicts_with_all = ["min_replicas", "max_replicas"])] num_replicas: Option, + /// Minimum number of replicas for horizontal autoscaling (requires the + /// horizontal autoscaling org feature). Mutually exclusive with --num-replicas. + #[arg(long, conflicts_with = "num_replicas")] + min_replicas: Option, + + /// Maximum number of replicas for horizontal autoscaling (requires the + /// horizontal autoscaling org feature). Mutually exclusive with --num-replicas. + #[arg(long, conflicts_with = "num_replicas")] + max_replicas: Option, + + /// Autoscaling mode: vertical (default) or horizontal. Omit to keep the + /// service's current mode. See `service create --autoscaling-mode`. + #[arg( + long, + value_parser = PossibleValuesParser::new( + clickhouse_cloud_api::models::AutoscalingMode::VALUES + ) + )] + autoscaling_mode: Option, + /// Allow scale to zero when idle #[arg(long)] idle_scaling: Option, @@ -1040,6 +1093,24 @@ pub enum ClickPipeCommands { command: ClickPipeSettingsCommands, }, + /// Discover a source schema without creating a pipe (beta) + #[command(after_help = "\\ +CONTEXT FOR AGENTS: + Infers the schema (column name + ClickHouse type) for a Kafka or Kinesis source + without creating a ClickPipe. Useful for filling in --column on `clickpipe create`. + Related: `clickhousectl cloud clickpipe create kafka|kinesis` to create a pipe with the discovered columns.")] + SchemaDiscover { + /// Service ID + service_id: String, + + #[command(subcommand)] + command: ClickPipeSchemaDiscoverCommands, + + /// Organization ID (auto-detected if not specified) + #[arg(long)] + org_id: Option, + }, + /// Create a ClickPipe Create { #[command(subcommand)] @@ -1047,6 +1118,15 @@ pub enum ClickPipeCommands { }, } +#[derive(Subcommand)] +pub enum ClickPipeSchemaDiscoverCommands { + /// Discover schema from a Kafka or Kafka-compatible source + Kafka(Box), + + /// Discover schema from an Amazon Kinesis stream + Kinesis(Box), +} + #[derive(Subcommand)] pub enum ClickPipeSettingsCommands { /// Get ClickPipe settings @@ -1193,6 +1273,16 @@ pub struct ObjectStorageCreateArgs { #[arg(long)] pub queue_url: Option, + /// Skip the initial load of existing objects and ingest only queue-notification + /// files. Only applicable when --queue-url is provided. + #[arg(long, requires = "queue_url")] + pub skip_initial_load: bool, + + /// Object key to start continuous ingestion after. Mutually exclusive with + /// --skip-initial-load (the API rejects both being set). + #[arg(long, conflicts_with = "skip_initial_load")] + pub start_after: Option, + /// CSV delimiter character (e.g., ",") #[arg(long)] pub delimiter: Option, @@ -1230,15 +1320,11 @@ pub struct ObjectStorageCreateArgs { pub org_id: Option, } +/// Source-connection fields for a Kafka / Kafka-compatible ClickPipe source. +/// Flattened into both `KafkaCreateArgs` (pipe creation) and the schema-discover +/// Kafka subcommand so the source field set has a single definition. #[derive(Args, Debug)] -pub struct KafkaCreateArgs { - /// Service ID - pub service_id: String, - - /// ClickPipe name - #[arg(long)] - pub name: String, - +pub struct KafkaSourceFields { /// Kafka broker(s) (e.g., "broker1:9092,broker2:9092") #[arg(long)] pub brokers: String, @@ -1251,18 +1337,6 @@ pub struct KafkaCreateArgs { #[arg(long, value_parser = PossibleValuesParser::new(KAFKA_FORMATS))] pub format: String, - /// Destination database - #[arg(long)] - pub database: String, - - /// Destination table - #[arg(long)] - pub table: String, - - /// Destination columns as name:type pairs (e.g., --column "event_id:Int64") - #[arg(long = "column")] - pub columns: Vec, - /// Kafka type #[arg( long, @@ -1342,14 +1416,10 @@ pub struct KafkaCreateArgs { /// Reverse private endpoint IDs (repeatable) #[arg(long = "reverse-private-endpoint-id")] pub reverse_private_endpoint_ids: Vec, - - /// Organization ID (auto-detected if not specified) - #[arg(long)] - pub org_id: Option, } #[derive(Args, Debug)] -pub struct KinesisCreateArgs { +pub struct KafkaCreateArgs { /// Service ID pub service_id: String, @@ -1357,17 +1427,8 @@ pub struct KinesisCreateArgs { #[arg(long)] pub name: String, - /// Kinesis stream name - #[arg(long)] - pub stream_name: String, - - /// AWS region (e.g., us-east-1) - #[arg(long)] - pub region: String, - - /// Data format - #[arg(long, value_parser = PossibleValuesParser::new(KINESIS_FORMATS))] - pub format: String, + #[command(flatten)] + pub source: KafkaSourceFields, /// Destination database #[arg(long)] @@ -1381,6 +1442,28 @@ pub struct KinesisCreateArgs { #[arg(long = "column")] pub columns: Vec, + /// Organization ID (auto-detected if not specified) + #[arg(long)] + pub org_id: Option, +} + +/// Source-connection fields for an Amazon Kinesis ClickPipe source. +/// Flattened into both `KinesisCreateArgs` (pipe creation) and the schema-discover +/// Kinesis subcommand so the source field set has a single definition. +#[derive(Args, Debug)] +pub struct KinesisSourceFields { + /// Kinesis stream name + #[arg(long)] + pub stream_name: String, + + /// AWS region (e.g., us-east-1) + #[arg(long)] + pub region: String, + + /// Data format + #[arg(long, value_parser = PossibleValuesParser::new(KINESIS_FORMATS))] + pub format: String, + /// Authentication #[arg( long, @@ -1416,6 +1499,31 @@ pub struct KinesisCreateArgs { /// Enable enhanced fan-out #[arg(long)] pub enhanced_fan_out: bool, +} + +#[derive(Args, Debug)] +pub struct KinesisCreateArgs { + /// Service ID + pub service_id: String, + + /// ClickPipe name + #[arg(long)] + pub name: String, + + #[command(flatten)] + pub source: KinesisSourceFields, + + /// Destination database + #[arg(long)] + pub database: String, + + /// Destination table + #[arg(long)] + pub table: String, + + /// Destination columns as name:type pairs (e.g., --column "event_id:Int64") + #[arg(long = "column")] + pub columns: Vec, /// Organization ID (auto-detected if not specified) #[arg(long)] @@ -1585,6 +1693,12 @@ pub struct MySqlCreateArgs { #[arg(long)] pub skip_cert_verification: bool, + /// Optional MySQL server_id the pipe declares itself as in the MySQL + /// replication topology (1-4294967295). Must be unique across replicas + /// connected to the source. If omitted, one is assigned automatically. + #[arg(long, value_parser = clap::value_parser!(u64).range(1..=4294967295))] + pub server_id: Option, + /// Organization ID (auto-detected if not specified) #[arg(long)] pub org_id: Option, @@ -1991,6 +2105,467 @@ mod tests { assert_eq!(enable_core_dumps, Some(true)); } + #[test] + fn parses_service_create_horizontal_autoscaling_flags() { + let cli = Cli::try_parse_from([ + "clickhousectl", + "cloud", + "service", + "create", + "--name", + "s", + "--min-replicas", + "2", + "--max-replicas", + "8", + "--autoscaling-mode", + "horizontal", + ]) + .unwrap(); + let Commands::Cloud(args) = cli.command else { + panic!("expected cloud command"); + }; + let CloudCommands::Service { command } = args.command else { + panic!("expected service command"); + }; + let ServiceCommands::Create { + min_replicas, + max_replicas, + autoscaling_mode, + num_replicas, + min_replica_memory_gb, + max_replica_memory_gb, + .. + } = command + else { + panic!("expected service create"); + }; + assert_eq!(min_replicas, Some(2)); + assert_eq!(max_replicas, Some(8)); + assert_eq!(autoscaling_mode.as_deref(), Some("horizontal")); + assert!(num_replicas.is_none()); + assert!(min_replica_memory_gb.is_none()); + assert!(max_replica_memory_gb.is_none()); + } + + #[test] + fn rejects_service_create_horizontal_vertical_mix() { + // --min-replicas conflicts with the vertical flags + let result = Cli::try_parse_from([ + "clickhousectl", + "cloud", + "service", + "create", + "--name", + "s", + "--min-replicas", + "2", + "--max-replicas", + "8", + "--num-replicas", + "3", + ]); + assert!(result.is_err()); + } + + #[test] + fn parses_service_create_horizontal_mode_with_memory_bounds() { + // Horizontal requires equal memory bounds, so the memory flags must + // combine with the mode and the replica band in one invocation. + let cli = Cli::try_parse_from([ + "clickhousectl", + "cloud", + "service", + "create", + "--name", + "s", + "--autoscaling-mode", + "horizontal", + "--min-replicas", + "2", + "--max-replicas", + "8", + "--min-replica-memory-gb", + "16", + "--max-replica-memory-gb", + "16", + ]) + .unwrap(); + let Commands::Cloud(args) = cli.command else { + panic!("expected cloud command"); + }; + let CloudCommands::Service { command } = args.command else { + panic!("expected service command"); + }; + let ServiceCommands::Create { + min_replicas, + max_replicas, + autoscaling_mode, + min_replica_memory_gb, + max_replica_memory_gb, + .. + } = command + else { + panic!("expected service create"); + }; + assert_eq!(min_replicas, Some(2)); + assert_eq!(max_replicas, Some(8)); + assert_eq!(autoscaling_mode.as_deref(), Some("horizontal")); + assert_eq!(min_replica_memory_gb, Some(16)); + assert_eq!(max_replica_memory_gb, Some(16)); + } + + #[test] + fn rejects_service_create_invalid_autoscaling_mode() { + let result = Cli::try_parse_from([ + "clickhousectl", + "cloud", + "service", + "create", + "--name", + "s", + "--min-replicas", + "2", + "--max-replicas", + "8", + "--autoscaling-mode", + "turbo", + ]); + assert!(result.is_err()); + } + + #[test] + fn parses_service_scale_horizontal_autoscaling_flags() { + let cli = Cli::try_parse_from([ + "clickhousectl", + "cloud", + "service", + "scale", + "svc-1", + "--min-replicas", + "2", + "--max-replicas", + "8", + "--autoscaling-mode", + "horizontal", + ]) + .unwrap(); + let Commands::Cloud(args) = cli.command else { + panic!("expected cloud command"); + }; + let CloudCommands::Service { command } = args.command else { + panic!("expected service command"); + }; + let ServiceCommands::Scale { + service_id, + min_replicas, + max_replicas, + autoscaling_mode, + num_replicas, + .. + } = command + else { + panic!("expected service scale"); + }; + assert_eq!(service_id, "svc-1"); + assert_eq!(min_replicas, Some(2)); + assert_eq!(max_replicas, Some(8)); + assert_eq!(autoscaling_mode.as_deref(), Some("horizontal")); + assert!(num_replicas.is_none()); + } + + #[test] + fn parses_service_scale_switch_to_vertical_in_one_call() { + // Switching a horizontal service back to vertical sends the mode and + // the target replica count in a single request. + let cli = Cli::try_parse_from([ + "clickhousectl", + "cloud", + "service", + "scale", + "svc-1", + "--autoscaling-mode", + "vertical", + "--num-replicas", + "3", + "--min-replica-memory-gb", + "8", + "--max-replica-memory-gb", + "32", + ]) + .unwrap(); + let Commands::Cloud(args) = cli.command else { + panic!("expected cloud command"); + }; + let CloudCommands::Service { command } = args.command else { + panic!("expected service command"); + }; + let ServiceCommands::Scale { + autoscaling_mode, + num_replicas, + min_replica_memory_gb, + max_replica_memory_gb, + min_replicas, + max_replicas, + .. + } = command + else { + panic!("expected service scale"); + }; + assert_eq!(autoscaling_mode.as_deref(), Some("vertical")); + assert_eq!(num_replicas, Some(3)); + assert_eq!(min_replica_memory_gb, Some(8)); + assert_eq!(max_replica_memory_gb, Some(32)); + assert!(min_replicas.is_none()); + assert!(max_replicas.is_none()); + } + + #[test] + fn rejects_service_scale_num_replicas_with_replica_band() { + let result = Cli::try_parse_from([ + "clickhousectl", + "cloud", + "service", + "scale", + "svc-1", + "--num-replicas", + "3", + "--min-replicas", + "2", + "--max-replicas", + "8", + ]); + assert!(result.is_err()); + } + + #[test] + fn parses_clickpipe_object_storage_ingestion_control_flags() { + let cli = Cli::try_parse_from([ + "clickhousectl", + "cloud", + "clickpipe", + "create", + "object-storage", + "svc-id", + "--name", + "t", + "--source-url", + "https://b.s3.us-east-1.amazonaws.com/d/*.json", + "--format", + "JSONEachRow", + "--database", + "d", + "--table", + "t", + "--column", + "id:Int64", + "--queue-url", + "https://sqs.us-east-1.amazonaws.com/123/q", + "--start-after", + "key1", + ]) + .unwrap(); + let Commands::Cloud(args) = cli.command else { + panic!("expected cloud command"); + }; + let CloudCommands::ClickPipe { command } = args.command else { + panic!("expected clickpipe command"); + }; + let ClickPipeCommands::Create { command } = *command else { + panic!("expected create"); + }; + let ClickPipeCreateCommands::ObjectStorage(args) = command else { + panic!("expected object-storage"); + }; + assert!(!args.skip_initial_load); + assert_eq!(args.start_after.as_deref(), Some("key1")); + assert_eq!( + args.queue_url.as_deref(), + Some("https://sqs.us-east-1.amazonaws.com/123/q") + ); + } + + #[test] + fn rejects_skip_initial_load_without_queue_url() { + let result = Cli::try_parse_from([ + "clickhousectl", + "cloud", + "clickpipe", + "create", + "object-storage", + "svc-id", + "--name", + "t", + "--source-url", + "https://b.s3.us-east-1.amazonaws.com/d/*.json", + "--format", + "JSONEachRow", + "--database", + "d", + "--table", + "t", + "--column", + "id:Int64", + "--skip-initial-load", + ]); + assert!(result.is_err()); + } + + #[test] + fn rejects_skip_initial_load_with_start_after() { + let result = Cli::try_parse_from([ + "clickhousectl", + "cloud", + "clickpipe", + "create", + "object-storage", + "svc-id", + "--name", + "t", + "--source-url", + "https://b.s3.us-east-1.amazonaws.com/d/*.json", + "--format", + "JSONEachRow", + "--database", + "d", + "--table", + "t", + "--column", + "id:Int64", + "--queue-url", + "https://sqs.us-east-1.amazonaws.com/123/q", + "--skip-initial-load", + "--start-after", + "key1", + ]); + assert!(result.is_err()); + } + + #[test] + fn parses_clickpipe_mysql_server_id() { + let cli = Cli::try_parse_from([ + "clickhousectl", + "cloud", + "clickpipe", + "create", + "mysql", + "svc-id", + "--name", + "t", + "--host", + "h", + "--username", + "u", + "--password", + "p", + "--table-mapping", + "db.t:t", + "--server-id", + "4242", + ]) + .unwrap(); + let Commands::Cloud(args) = cli.command else { + panic!("expected cloud command"); + }; + let CloudCommands::ClickPipe { command } = args.command else { + panic!("expected clickpipe command"); + }; + let ClickPipeCommands::Create { command } = *command else { + panic!("expected create"); + }; + let ClickPipeCreateCommands::MySQL(args) = command else { + panic!("expected mysql"); + }; + assert_eq!(args.server_id, Some(4242)); + } + + #[test] + fn rejects_clickpipe_mysql_server_id_out_of_range() { + // 0 is below the minimum (1) + let result = Cli::try_parse_from([ + "clickhousectl", + "cloud", + "clickpipe", + "create", + "mysql", + "svc-id", + "--name", + "t", + "--host", + "h", + "--username", + "u", + "--password", + "p", + "--table-mapping", + "db.t:t", + "--server-id", + "0", + ]); + assert!(result.is_err()); + + // 4294967296 is above the maximum (4294967295) + let result = Cli::try_parse_from([ + "clickhousectl", + "cloud", + "clickpipe", + "create", + "mysql", + "svc-id", + "--name", + "t", + "--host", + "h", + "--username", + "u", + "--password", + "p", + "--table-mapping", + "db.t:t", + "--server-id", + "4294967296", + ]); + assert!(result.is_err()); + } + + #[test] + fn parses_clickpipe_schema_discover_kafka() { + let cli = Cli::try_parse_from([ + "clickhousectl", + "cloud", + "clickpipe", + "schema-discover", + "svc-1", + "kafka", + "--brokers", + "b:9092", + "--topics", + "t", + "--format", + "JSONEachRow", + ]) + .unwrap(); + let Commands::Cloud(args) = cli.command else { + panic!("expected cloud command"); + }; + let CloudCommands::ClickPipe { command } = args.command else { + panic!("expected clickpipe command"); + }; + let ClickPipeCommands::SchemaDiscover { + service_id, + command, + .. + } = *command + else { + panic!("expected schema-discover"); + }; + assert_eq!(service_id, "svc-1"); + assert!(matches!( + command, + ClickPipeSchemaDiscoverCommands::Kafka(_) + )); + } + #[test] fn parses_private_endpoint_config_and_password_hash_flags() { let cli = Cli::try_parse_from([ @@ -2479,10 +3054,31 @@ mod tests { assert_write(&["clickhousectl", "cloud", "postgres", "get", "pg-1"], false); assert_write(&["clickhousectl", "cloud", "postgres", "certs", "get", "pg-1"], false); assert_write(&["clickhousectl", "cloud", "postgres", "config", "get", "pg-1"], false); + } #[test] fn is_write_command_destructive_commands() { + // ClickPipe schema discovery is side-effect-free, but the API gateway + // rejects OAuth/JWT on the endpoint, so it requires API-key auth. + assert_write( + &[ + "clickhousectl", + "cloud", + "clickpipe", + "schema-discover", + "svc-1", + "kafka", + "--brokers", + "b:9092", + "--topics", + "t", + "--format", + "JSONEachRow", + ], + true, + ); + // Org write assert_write( &[ diff --git a/crates/clickhousectl/src/cloud/client.rs b/crates/clickhousectl/src/cloud/client.rs index 5928da3..5896717 100644 --- a/crates/clickhousectl/src/cloud/client.rs +++ b/crates/clickhousectl/src/cloud/client.rs @@ -1065,6 +1065,20 @@ impl CloudClient { Self::unwrap_response(response) } + pub async fn click_pipe_schema_discovery( + &self, + org_id: &str, + service_id: &str, + request: &clickhouse_cloud_api::models::ClickPipeSchemaDiscoveryRequest, + ) -> Result { + let response = self + .api() + .click_pipe_schema_discovery(org_id, service_id, request) + .await + .map_err(|e| self.convert_error(e))?; + Self::unwrap_response(response) + } + // Helper to get the default organization pub async fn get_default_org_id(&self) -> Result { let orgs = self.list_organizations().await?; diff --git a/crates/clickhousectl/src/cloud/commands.rs b/crates/clickhousectl/src/cloud/commands.rs index b87aa5f..c71e8fb 100644 --- a/crates/clickhousectl/src/cloud/commands.rs +++ b/crates/clickhousectl/src/cloud/commands.rs @@ -13,7 +13,7 @@ use clickhouse_cloud_api::models::{ ServicePostRequest, ServicePostRequestCompliancetype, ServicePostRequestProfile, ServicePostRequestProvider, ServicePostRequestRegion, ServicePostRequestReleasechannel, ServiceReplicaScalingPatchRequest, - ServiceStatePatchRequestCommand, ServicPrivateEndpointePostRequest, + ServiceStatePatchRequestCommand, ServicPrivateEndpointePostRequest, AutoscalingMode, }; use std::io::{IsTerminal, Write}; use tabled::{Table, Tabled, settings::Style}; @@ -492,6 +492,9 @@ pub struct CreateServiceOptions { pub min_replica_memory_gb: Option, pub max_replica_memory_gb: Option, pub num_replicas: Option, + pub min_replicas: Option, + pub max_replicas: Option, + pub autoscaling_mode: Option, pub idle_scaling: Option, pub idle_timeout_minutes: Option, pub ip_allow: Vec, @@ -582,6 +585,50 @@ pub struct BackupConfigUpdateOptions { pub org_id: Option, } +/// Resolved horizontal-autoscaling fields for a service create/scale request. +struct HorizontalAutoscaling { + autoscaling_mode: Option, + min_replicas: Option, + max_replicas: Option, +} + +/// Resolve the horizontal-autoscaling fields shared by `service create` and +/// `service scale`. +/// +/// The mode is sent only when `--autoscaling-mode` is given explicitly. The +/// API resolves an omitted mode itself, and a min/max band with the mode +/// omitted and min == max is accepted as a vertical fixed replica count that +/// needs no horizontal entitlement — inferring `horizontal` here would change +/// those semantics. +/// +/// Rejects `--min-replicas` without `--max-replicas` (and vice versa) with a +/// clear error before any network call. clap already rejects mixing the +/// horizontal pair with `--num-replicas`; the memory flags and +/// `--autoscaling-mode` combine freely with either set because a single +/// request can switch modes (e.g. `--autoscaling-mode vertical +/// --num-replicas 3`, or `--autoscaling-mode horizontal` with the equal +/// memory bounds horizontal requires). Remaining combination rules are the +/// API's to enforce. +fn resolve_horizontal_autoscaling( + autoscaling_mode: Option<&str>, + min_replicas: Option, + max_replicas: Option, +) -> Result> { + if min_replicas.is_some() != max_replicas.is_some() { + return Err("--min-replicas and --max-replicas must be specified together".into()); + } + let autoscaling_mode = autoscaling_mode + .map(|value| { + parse_serde_enum::(value, "autoscaling_mode", AutoscalingMode::VALUES) + }) + .transpose()?; + Ok(HorizontalAutoscaling { + autoscaling_mode, + min_replicas: min_replicas.map(f64::from), + max_replicas: max_replicas.map(f64::from), + }) +} + fn build_create_service_request( opts: &CreateServiceOptions, ) -> Result> { @@ -594,6 +641,12 @@ fn build_create_service_request( parse_ip_access_entries(&opts.ip_allow).unwrap_or_default() }; + let horizontal = resolve_horizontal_autoscaling( + opts.autoscaling_mode.as_deref(), + opts.min_replicas, + opts.max_replicas, + )?; + Ok(ServicePostRequest { name: opts.name.clone(), provider: parse_serde_enum::( @@ -656,10 +709,10 @@ fn build_create_service_request( endpoints: parse_service_endpoint_changes(&opts.enable_endpoints, &opts.disable_endpoints)?, enable_core_dumps: opts.enable_core_dumps, // Fields not exposed in CLI - autoscaling_mode: None, + autoscaling_mode: horizontal.autoscaling_mode, byoc_id: None, - min_replicas: None, - max_replicas: None, + min_replicas: horizontal.min_replicas, + max_replicas: horizontal.max_replicas, // Deprecated fields — only exist (and stay None) under the // `deprecated-fields` feature; gated out of the struct otherwise. #[cfg(feature = "deprecated-fields")] @@ -1000,9 +1053,8 @@ pub async fn clickpipe_create_s3( azure_container_name: args.azure_container_name.clone(), path: args.path.clone(), service_account_key, - // Not exposed as CLI flags yet - skip_initial_load: None, - start_after: None, + skip_initial_load: if args.skip_initial_load { Some(true) } else { None }, + start_after: args.start_after.clone(), }; let request = ClickPipePostRequest { @@ -1032,7 +1084,7 @@ pub async fn clickpipe_create_s3( /// stays pure and testable. fn build_kafka_credentials( authentication: &clickhouse_cloud_api::models::ClickPipePostKafkaSourceAuthentication, - args: &crate::cloud::cli::KafkaCreateArgs, + args: &crate::cloud::cli::KafkaSourceFields, mtls_contents: Option<(String, String)>, ) -> Result { use clickhouse_cloud_api::models::ClickPipePostKafkaSourceAuthentication as Auth; @@ -1067,21 +1119,20 @@ fn build_kafka_credentials( } } -pub async fn clickpipe_create_kafka( - client: &CloudClient, - args: &crate::cloud::cli::KafkaCreateArgs, - json: bool, -) -> Result<(), Box> { +/// Build a `ClickPipePostKafkaSource` from the CLI args, performing all +/// authentication/credential/schema-registry/CA validation up front so bad +/// invocations fail fast before any network call. Shared by the +/// `clickpipe create kafka` and `clickpipe schema-discover kafka` +/// handlers. +fn build_kafka_source( + args: &crate::cloud::cli::KafkaSourceFields, +) -> Result> { use clickhouse_cloud_api::models::{ ClickPipeKafkaOffset, ClickPipeKafkaSchemaRegistryCredentials, ClickPipeMutateKafkaSchemaRegistry, ClickPipePostKafkaSource, - ClickPipePostKafkaSourceAuthentication, ClickPipePostRequest, ClickPipePostSource, + ClickPipePostKafkaSourceAuthentication, }; - // Validate args and build credentials before any network call so bad - // invocations fail fast. - let parsed_columns = parse_columns(&args.columns)?; - let authentication: ClickPipePostKafkaSourceAuthentication = match args.auth.as_deref() { Some(a) => parse_enum(a)?, None => ClickPipePostKafkaSourceAuthentication::default(), @@ -1134,7 +1185,7 @@ pub async fn clickpipe_create_kafka( None => None, }; - let source = ClickPipePostKafkaSource { + Ok(ClickPipePostKafkaSource { r#type: parse_enum(&args.kafka_type)?, format: parse_enum(&args.format)?, brokers: args.brokers.clone(), @@ -1151,8 +1202,56 @@ pub async fn clickpipe_create_kafka( schema_registry, ca_certificate: ca_cert_contents, reverse_private_endpoint_ids: args.reverse_private_endpoint_ids.clone(), + }) +} + +/// Build a `ClickPipePostKinesisSource` from the CLI args. Shared by the +/// `clickpipe create kinesis` and `clickpipe schema-discover kinesis` +/// handlers. +fn build_kinesis_source( + args: &crate::cloud::cli::KinesisSourceFields, +) -> Result> { + use clickhouse_cloud_api::models::{ClickPipePostKinesisSource, MskIamUser}; + + let access_key = match (args.access_key_id.as_deref(), args.secret_key.as_deref()) { + (Some(k), Some(s)) => Some(MskIamUser { + access_key_id: k.to_string(), + secret_key: s.to_string(), + }), + _ => None, }; + Ok(ClickPipePostKinesisSource { + format: parse_enum(&args.format)?, + stream_name: args.stream_name.clone(), + region: args.region.clone(), + authentication: parse_enum(&args.auth)?, + iam_role: args.iam_role.clone(), + access_key, + use_enhanced_fan_out: if args.enhanced_fan_out { Some(true) } else { None }, + iterator_type: parse_enum(&args.iterator_type)?, + timestamp: args + .iterator_timestamp + .map(|t| { + i64::try_from(t) + .map_err(|_| format!("--iterator-timestamp {t} is out of range")) + }) + .transpose()?, + }) +} + +pub async fn clickpipe_create_kafka( + client: &CloudClient, + args: &crate::cloud::cli::KafkaCreateArgs, + json: bool, +) -> Result<(), Box> { + use clickhouse_cloud_api::models::{ClickPipePostRequest, ClickPipePostSource}; + + // Validate args and build the source before any network call so bad + // invocations fail fast. + let parsed_columns = parse_columns(&args.columns)?; + let source = build_kafka_source(&args.source)?; + let request = ClickPipePostRequest { name: args.name.clone(), source: ClickPipePostSource { @@ -1176,32 +1275,11 @@ pub async fn clickpipe_create_kinesis( args: &crate::cloud::cli::KinesisCreateArgs, json: bool, ) -> Result<(), Box> { - use clickhouse_cloud_api::models::{ - ClickPipePostKinesisSource, ClickPipePostRequest, ClickPipePostSource, MskIamUser, - }; + use clickhouse_cloud_api::models::{ClickPipePostRequest, ClickPipePostSource}; let org_id = resolve_org_id(client, args.org_id.as_deref()).await?; let parsed_columns = parse_columns(&args.columns)?; - - let access_key = match (args.access_key_id.as_deref(), args.secret_key.as_deref()) { - (Some(k), Some(s)) => Some(MskIamUser { - access_key_id: k.to_string(), - secret_key: s.to_string(), - }), - _ => None, - }; - - let source = ClickPipePostKinesisSource { - format: parse_enum(&args.format)?, - stream_name: args.stream_name.clone(), - region: args.region.clone(), - authentication: parse_enum(&args.auth)?, - iam_role: args.iam_role.clone(), - access_key, - use_enhanced_fan_out: if args.enhanced_fan_out { Some(true) } else { None }, - iterator_type: parse_enum(&args.iterator_type)?, - timestamp: args.iterator_timestamp.map(|t| t as i64), - }; + let source = build_kinesis_source(&args.source)?; let request = ClickPipePostRequest { name: args.name.clone(), @@ -1220,6 +1298,76 @@ pub async fn clickpipe_create_kinesis( Ok(()) } +/// Discover the inferred schema for a Kafka or Kinesis source without creating +/// a ClickPipe (Beta). Side-effect-free, but the API gateway rejects +/// OAuth/Bearer on this POST endpoint, so it is classified as a write command +/// and requires API key auth. +pub async fn clickpipe_schema_discover( + client: &CloudClient, + service_id: &str, + command: &crate::cloud::cli::ClickPipeSchemaDiscoverCommands, + org_id: Option<&str>, + json: bool, +) -> Result<(), Box> { + use clickhouse_cloud_api::models::{ + ClickPipeSchemaDiscoveryRequest, ClickPipeSchemaDiscoverySource, + }; + + let source = match command { + crate::cloud::cli::ClickPipeSchemaDiscoverCommands::Kafka(args) => { + ClickPipeSchemaDiscoverySource { + kafka: Some(build_kafka_source(args)?), + kinesis: None, + } + } + crate::cloud::cli::ClickPipeSchemaDiscoverCommands::Kinesis(args) => { + ClickPipeSchemaDiscoverySource { + kafka: None, + kinesis: Some(build_kinesis_source(args)?), + } + } + }; + + let request = ClickPipeSchemaDiscoveryRequest { source }; + let org_id = resolve_org_id(client, org_id).await?; + let response = client + .click_pipe_schema_discovery(&org_id, service_id, &request) + .await?; + + if json { + println!("{}", serde_json::to_string_pretty(&response)?); + } else { + #[derive(Tabled)] + struct Row { + #[tabled(rename = "Name")] + name: String, + #[tabled(rename = "Type")] + r#type: String, + #[tabled(rename = "Optional")] + optional: String, + } + let rows: Vec = response + .fields + .into_iter() + .map(|f| Row { + name: f.name, + r#type: f.r#type, + optional: match f.optional { + Some(true) => "true".to_string(), + Some(false) => "false".to_string(), + None => "".to_string(), + }, + }) + .collect(); + if rows.is_empty() { + println!("No fields discovered"); + } else { + println!("{}", Table::new(rows).with(Style::markdown())); + } + } + Ok(()) +} + pub async fn clickpipe_get( client: &CloudClient, service_id: &str, @@ -1614,7 +1762,7 @@ pub async fn clickpipe_create_mysql( } else { None }, - server_id: None, + server_id: args.server_id.map(|v| v as i64), settings: ClickPipeMySQLPipeSettings { replication_mode: parse_enum(&args.replication_mode)?, replication_mechanism: Some(parse_enum(&args.replication_mechanism)?), @@ -1895,34 +2043,49 @@ pub async fn service_update( Ok(()) } +#[derive(Default)] pub struct ServiceScaleOptions { pub min_replica_memory_gb: Option, pub max_replica_memory_gb: Option, pub num_replicas: Option, + pub min_replicas: Option, + pub max_replicas: Option, + pub autoscaling_mode: Option, pub idle_scaling: Option, pub idle_timeout_minutes: Option, pub org_id: Option, } +fn build_service_scale_request( + opts: &ServiceScaleOptions, +) -> Result> { + let horizontal = resolve_horizontal_autoscaling( + opts.autoscaling_mode.as_deref(), + opts.min_replicas, + opts.max_replicas, + )?; + + Ok(ServiceReplicaScalingPatchRequest { + autoscaling_mode: horizontal.autoscaling_mode, + min_replica_memory_gb: opts.min_replica_memory_gb.map(f64::from), + max_replica_memory_gb: opts.max_replica_memory_gb.map(f64::from), + min_replicas: horizontal.min_replicas, + max_replicas: horizontal.max_replicas, + num_replicas: opts.num_replicas.map(f64::from), + idle_scaling: opts.idle_scaling, + idle_timeout_minutes: opts.idle_timeout_minutes.map(f64::from), + }) +} + pub async fn service_scale( client: &CloudClient, service_id: &str, opts: ServiceScaleOptions, json: bool, ) -> Result<(), Box> { + let request = build_service_scale_request(&opts)?; let org_id = resolve_org_id(client, opts.org_id.as_deref()).await?; - let request = ServiceReplicaScalingPatchRequest { - min_replica_memory_gb: opts.min_replica_memory_gb.map(f64::from), - max_replica_memory_gb: opts.max_replica_memory_gb.map(f64::from), - min_replicas: None, - max_replicas: None, - num_replicas: opts.num_replicas.map(f64::from), - idle_scaling: opts.idle_scaling, - idle_timeout_minutes: opts.idle_timeout_minutes.map(f64::from), - ..Default::default() - }; - let svc = client .update_replica_scaling(&org_id, service_id, &request) .await?; @@ -1931,9 +2094,21 @@ pub async fn service_scale( println!("{}", serde_json::to_string_pretty(&svc)?); } else { println!("Service {} scaling updated", svc.name); - println!(" Min Memory/Replica: {} GB", svc.min_replica_memory_gb); - println!(" Max Memory/Replica: {} GB", svc.max_replica_memory_gb); - println!(" Replicas: {}", svc.num_replicas); + println!(" Autoscaling Mode: {}", svc.autoscaling_mode); + match svc.autoscaling_mode { + AutoscalingMode::Horizontal => { + println!(" Min Replicas: {}", svc.min_replicas); + println!(" Max Replicas: {}", svc.max_replicas); + println!(" Memory/Replica: {} GB", svc.replica_memory_gb); + } + AutoscalingMode::Vertical => { + println!(" Min Memory/Replica: {} GB", svc.min_replica_memory_gb); + println!(" Max Memory/Replica: {} GB", svc.max_replica_memory_gb); + println!(" Replicas: {}", svc.num_replicas); + } + // A mode this CLI version doesn't know; don't guess which fields apply. + _ => {} + } } Ok(()) } @@ -2856,6 +3031,9 @@ mod tests { min_replica_memory_gb: Some(24), max_replica_memory_gb: Some(48), num_replicas: Some(3), + min_replicas: None, + max_replicas: None, + autoscaling_mode: None, idle_scaling: Some(true), idle_timeout_minutes: Some(10), ip_allow: vec!["10.0.0.0/8".to_string()], @@ -2895,6 +3073,9 @@ mod tests { min_replica_memory_gb: None, max_replica_memory_gb: None, num_replicas: None, + min_replicas: None, + max_replicas: None, + autoscaling_mode: None, idle_scaling: None, idle_timeout_minutes: None, ip_allow: vec![], @@ -2930,6 +3111,9 @@ mod tests { min_replica_memory_gb: None, max_replica_memory_gb: None, num_replicas: None, + min_replicas: None, + max_replicas: None, + autoscaling_mode: None, idle_scaling: None, idle_timeout_minutes: None, ip_allow: vec![], @@ -3163,6 +3347,235 @@ mod tests { ); } + #[test] + fn build_create_service_request_horizontal_autoscaling_on_wire() { + // Maximal: explicit --autoscaling-mode horizontal + min/max replicas. + let opts = CreateServiceOptions { + name: "svc".to_string(), + provider: "aws".to_string(), + region: "us-east-1".to_string(), + min_replicas: Some(2), + max_replicas: Some(8), + autoscaling_mode: Some("horizontal".to_string()), + ..Default::default() + }; + let request = build_create_service_request(&opts).unwrap(); + let json = serde_json::to_value(&request).unwrap(); + assert_eq!(json["autoscalingMode"], "horizontal"); + assert_eq!(json["minReplicas"], 2.0); + assert_eq!(json["maxReplicas"], 8.0); + // Vertical fields stay absent. + assert!(json.get("numReplicas").is_none()); + assert!(json.get("minReplicaMemoryGb").is_none()); + assert!(json.get("maxReplicaMemoryGb").is_none()); + } + + #[test] + fn build_create_service_request_replica_pair_without_mode_omits_mode() { + // No explicit --autoscaling-mode: the replica pair passes through with + // the mode absent. The API resolves an omitted mode itself — an equal + // band is accepted as a vertical fixed replica count without the + // horizontal entitlement, so the CLI must not inject "horizontal". + let opts = CreateServiceOptions { + name: "svc".to_string(), + provider: "aws".to_string(), + region: "us-east-1".to_string(), + min_replicas: Some(1), + max_replicas: Some(4), + ..Default::default() + }; + let request = build_create_service_request(&opts).unwrap(); + let json = serde_json::to_value(&request).unwrap(); + assert!(json.get("autoscalingMode").is_none()); + assert_eq!(json["minReplicas"], 1.0); + assert_eq!(json["maxReplicas"], 4.0); + } + + #[test] + fn build_create_service_request_vertical_omits_horizontal_fields() { + // Minimal: vertical-only usage leaves autoscalingMode/replicas absent. + let opts = CreateServiceOptions { + name: "svc".to_string(), + provider: "aws".to_string(), + region: "us-east-1".to_string(), + num_replicas: Some(3), + min_replica_memory_gb: Some(24), + max_replica_memory_gb: Some(48), + ..Default::default() + }; + let request = build_create_service_request(&opts).unwrap(); + let json = serde_json::to_value(&request).unwrap(); + assert!(json.get("autoscalingMode").is_none()); + assert!(json.get("minReplicas").is_none()); + assert!(json.get("maxReplicas").is_none()); + assert_eq!(json["numReplicas"], 3.0); + } + + #[test] + fn build_create_service_request_rejects_min_without_max_replicas() { + let opts = CreateServiceOptions { + name: "svc".to_string(), + provider: "aws".to_string(), + region: "us-east-1".to_string(), + min_replicas: Some(2), + ..Default::default() + }; + let err = build_create_service_request(&opts).unwrap_err(); + assert!( + err.to_string().contains("--min-replicas"), + "error should guide the user: {}", + err + ); + + let opts = CreateServiceOptions { + name: "svc".to_string(), + provider: "aws".to_string(), + region: "us-east-1".to_string(), + max_replicas: Some(8), + ..Default::default() + }; + let err = build_create_service_request(&opts).unwrap_err(); + assert!( + err.to_string().contains("--min-replicas"), + "error should guide the user: {}", + err + ); + } + + #[test] + fn build_create_service_request_rejects_invalid_autoscaling_mode() { + let opts = CreateServiceOptions { + name: "svc".to_string(), + provider: "aws".to_string(), + region: "us-east-1".to_string(), + min_replicas: Some(2), + max_replicas: Some(8), + autoscaling_mode: Some("turbo".to_string()), + ..Default::default() + }; + let err = build_create_service_request(&opts).unwrap_err(); + assert!( + err.to_string().contains("turbo"), + "error should mention the bad value: {}", + err + ); + } + + #[test] + fn resolve_horizontal_autoscaling_explicit_vertical_with_no_replicas() { + // Explicit --autoscaling-mode vertical with no replica pair → mode set, + // replicas absent (lets the server apply vertical defaults). + let resolved = + resolve_horizontal_autoscaling(Some("vertical"), None, None).unwrap(); + assert_eq!(resolved.autoscaling_mode, Some(AutoscalingMode::Vertical)); + assert!(resolved.min_replicas.is_none()); + assert!(resolved.max_replicas.is_none()); + } + + #[test] + fn build_service_scale_request_horizontal_autoscaling_on_wire() { + let opts = ServiceScaleOptions { + min_replicas: Some(2), + max_replicas: Some(8), + autoscaling_mode: Some("horizontal".to_string()), + ..Default::default() + }; + let request = build_service_scale_request(&opts).unwrap(); + let json = serde_json::to_value(&request).unwrap(); + assert_eq!(json["autoscalingMode"], "horizontal"); + assert_eq!(json["minReplicas"], 2.0); + assert_eq!(json["maxReplicas"], 8.0); + assert!(json.get("numReplicas").is_none()); + assert!(json.get("minReplicaMemoryGb").is_none()); + assert!(json.get("maxReplicaMemoryGb").is_none()); + } + + #[test] + fn build_service_scale_request_switch_to_vertical_on_wire() { + // Switching a horizontal service back to vertical sends the mode and + // the vertical fields in one request. + let opts = ServiceScaleOptions { + autoscaling_mode: Some("vertical".to_string()), + num_replicas: Some(3), + min_replica_memory_gb: Some(8), + max_replica_memory_gb: Some(32), + ..Default::default() + }; + let request = build_service_scale_request(&opts).unwrap(); + let json = serde_json::to_value(&request).unwrap(); + assert_eq!(json["autoscalingMode"], "vertical"); + assert_eq!(json["numReplicas"], 3.0); + assert_eq!(json["minReplicaMemoryGb"], 8.0); + assert_eq!(json["maxReplicaMemoryGb"], 32.0); + assert!(json.get("minReplicas").is_none()); + assert!(json.get("maxReplicas").is_none()); + } + + #[test] + fn build_service_scale_request_switch_to_horizontal_with_memory_on_wire() { + // Switching to horizontal pins the equal per-replica memory the mode + // requires in the same request. + let opts = ServiceScaleOptions { + autoscaling_mode: Some("horizontal".to_string()), + min_replicas: Some(2), + max_replicas: Some(8), + min_replica_memory_gb: Some(16), + max_replica_memory_gb: Some(16), + ..Default::default() + }; + let request = build_service_scale_request(&opts).unwrap(); + let json = serde_json::to_value(&request).unwrap(); + assert_eq!(json["autoscalingMode"], "horizontal"); + assert_eq!(json["minReplicas"], 2.0); + assert_eq!(json["maxReplicas"], 8.0); + assert_eq!(json["minReplicaMemoryGb"], 16.0); + assert_eq!(json["maxReplicaMemoryGb"], 16.0); + assert!(json.get("numReplicas").is_none()); + } + + #[test] + fn build_service_scale_request_rejects_min_without_max_replicas() { + let opts = ServiceScaleOptions { + max_replicas: Some(8), + ..Default::default() + }; + let err = build_service_scale_request(&opts).unwrap_err(); + assert!( + err.to_string().contains("--min-replicas"), + "error should guide the user: {}", + err + ); + } + + #[test] + fn build_kinesis_source_rejects_out_of_range_iterator_timestamp() { + let args = crate::cloud::cli::KinesisSourceFields { + stream_name: "stream".to_string(), + region: "us-east-1".to_string(), + format: "JSONEachRow".to_string(), + auth: "IAM_ROLE".to_string(), + iam_role: None, + access_key_id: None, + secret_key: None, + iterator_type: "AT_TIMESTAMP".to_string(), + iterator_timestamp: Some(u64::MAX), + enhanced_fan_out: false, + }; + let err = build_kinesis_source(&args).unwrap_err(); + assert!( + err.to_string().contains("out of range"), + "error should mention the range: {}", + err + ); + + let args = crate::cloud::cli::KinesisSourceFields { + iterator_timestamp: Some(1_750_000_000), + ..args + }; + let source = build_kinesis_source(&args).unwrap(); + assert_eq!(source.timestamp, Some(1_750_000_000)); + } + #[test] fn build_update_service_request_rejects_invalid_release_channel() { let opts = ServiceUpdateOptions { @@ -3351,30 +3764,32 @@ mod tests { crate::cloud::cli::KafkaCreateArgs { service_id: "svc".into(), name: "pipe".into(), - brokers: "b:9092".into(), - topics: "t".into(), - format: "JSONEachRow".into(), + source: crate::cloud::cli::KafkaSourceFields { + brokers: "b:9092".into(), + topics: "t".into(), + format: "JSONEachRow".into(), + kafka_type: "kafka".into(), + consumer_group: None, + auth: None, + username: None, + password: None, + iam_role: None, + access_key_id: None, + secret_key: None, + offset: "from_beginning".into(), + offset_timestamp: None, + schema_registry_url: None, + schema_registry_username: None, + schema_registry_password: None, + ca_certificate: None, + client_certificate: None, + client_key: None, + schema_registry_ca_certificate: None, + reverse_private_endpoint_ids: vec![], + }, database: "d".into(), table: "t".into(), columns: vec![], - kafka_type: "kafka".into(), - consumer_group: None, - auth: None, - username: None, - password: None, - iam_role: None, - access_key_id: None, - secret_key: None, - offset: "from_beginning".into(), - offset_timestamp: None, - schema_registry_url: None, - schema_registry_username: None, - schema_registry_password: None, - ca_certificate: None, - client_certificate: None, - client_key: None, - schema_registry_ca_certificate: None, - reverse_private_endpoint_ids: vec![], org_id: None, } } @@ -3383,10 +3798,10 @@ mod tests { fn kafka_credentials_plain_shape() { use clickhouse_cloud_api::models::ClickPipePostKafkaSourceAuthentication as Auth; let mut args = kafka_args(); - args.auth = Some("PLAIN".into()); - args.username = Some("u".into()); - args.password = Some("p".into()); - let creds = super::build_kafka_credentials(&Auth::PLAIN, &args, None).unwrap(); + args.source.auth = Some("PLAIN".into()); + args.source.username = Some("u".into()); + args.source.password = Some("p".into()); + let creds = super::build_kafka_credentials(&Auth::PLAIN, &args.source, None).unwrap(); assert_eq!(creds["username"], "u"); assert_eq!(creds["password"], "p"); } @@ -3395,10 +3810,10 @@ mod tests { fn kafka_credentials_iam_user_shape() { use clickhouse_cloud_api::models::ClickPipePostKafkaSourceAuthentication as Auth; let mut args = kafka_args(); - args.auth = Some("IAM_USER".into()); - args.access_key_id = Some("AKIA".into()); - args.secret_key = Some("secret".into()); - let creds = super::build_kafka_credentials(&Auth::IAM_USER, &args, None).unwrap(); + args.source.auth = Some("IAM_USER".into()); + args.source.access_key_id = Some("AKIA".into()); + args.source.secret_key = Some("secret".into()); + let creds = super::build_kafka_credentials(&Auth::IAM_USER, &args.source, None).unwrap(); // MskIamUser wire shape is {accessKeyId, secretKey} — NOT snake_case. assert_eq!(creds["accessKeyId"], "AKIA"); assert_eq!(creds["secretKey"], "secret"); @@ -3409,11 +3824,11 @@ mod tests { fn kafka_credentials_iam_role_is_null() { use clickhouse_cloud_api::models::ClickPipePostKafkaSourceAuthentication as Auth; let mut args = kafka_args(); - args.auth = Some("IAM_ROLE".into()); - args.iam_role = Some("arn:aws:iam::123:role/Foo".into()); + args.source.auth = Some("IAM_ROLE".into()); + args.source.iam_role = Some("arn:aws:iam::123:role/Foo".into()); // IAM_ROLE sends credentials=null; the role ARN flows through the // top-level `iamRole` field on the Kafka source, not credentials. - let creds = super::build_kafka_credentials(&Auth::IAM_ROLE, &args, None).unwrap(); + let creds = super::build_kafka_credentials(&Auth::IAM_ROLE, &args.source, None).unwrap(); assert!(creds.is_null()); } @@ -3422,7 +3837,7 @@ mod tests { use clickhouse_cloud_api::models::ClickPipePostKafkaSourceAuthentication as Auth; let args = kafka_args(); let contents = Some(("CERT_PEM".into(), "KEY_PEM".into())); - let creds = super::build_kafka_credentials(&Auth::MUTUAL_TLS, &args, contents).unwrap(); + let creds = super::build_kafka_credentials(&Auth::MUTUAL_TLS, &args.source, contents).unwrap(); assert_eq!(creds["certificate"], "CERT_PEM"); assert_eq!(creds["privateKey"], "KEY_PEM"); } @@ -3431,7 +3846,7 @@ mod tests { fn kafka_credentials_iam_user_missing_args_errors() { use clickhouse_cloud_api::models::ClickPipePostKafkaSourceAuthentication as Auth; let args = kafka_args(); - let err = super::build_kafka_credentials(&Auth::IAM_USER, &args, None).unwrap_err(); + let err = super::build_kafka_credentials(&Auth::IAM_USER, &args.source, None).unwrap_err(); assert!(err.contains("--access-key-id")); } @@ -3439,8 +3854,8 @@ mod tests { fn kafka_credentials_iam_role_missing_arn_errors() { use clickhouse_cloud_api::models::ClickPipePostKafkaSourceAuthentication as Auth; let mut args = kafka_args(); - args.auth = Some("IAM_ROLE".into()); - let err = super::build_kafka_credentials(&Auth::IAM_ROLE, &args, None).unwrap_err(); + args.source.auth = Some("IAM_ROLE".into()); + let err = super::build_kafka_credentials(&Auth::IAM_ROLE, &args.source, None).unwrap_err(); assert!(err.contains("--iam-role")); } } diff --git a/crates/clickhousectl/src/main.rs b/crates/clickhousectl/src/main.rs index 0abde89..d2aefea 100644 --- a/crates/clickhousectl/src/main.rs +++ b/crates/clickhousectl/src/main.rs @@ -469,6 +469,9 @@ async fn run_cloud(args: CloudArgs) -> Result<()> { min_replica_memory_gb, max_replica_memory_gb, num_replicas, + min_replicas, + max_replicas, + autoscaling_mode, idle_scaling, idle_timeout_minutes, ip_allow, @@ -495,6 +498,9 @@ async fn run_cloud(args: CloudArgs) -> Result<()> { min_replica_memory_gb, max_replica_memory_gb, num_replicas, + min_replicas, + max_replicas, + autoscaling_mode, idle_scaling, idle_timeout_minutes, ip_allow, @@ -574,6 +580,9 @@ async fn run_cloud(args: CloudArgs) -> Result<()> { min_replica_memory_gb, max_replica_memory_gb, num_replicas, + min_replicas, + max_replicas, + autoscaling_mode, idle_scaling, idle_timeout_minutes, org_id, @@ -585,6 +594,9 @@ async fn run_cloud(args: CloudArgs) -> Result<()> { min_replica_memory_gb, max_replica_memory_gb, num_replicas, + min_replicas, + max_replicas, + autoscaling_mode, idle_scaling, idle_timeout_minutes, org_id, @@ -1023,6 +1035,20 @@ async fn run_cloud(args: CloudArgs) -> Result<()> { .await } }, + ClickPipeCommands::SchemaDiscover { + service_id, + command, + org_id, + } => { + cloud::commands::clickpipe_schema_discover( + &client, + &service_id, + &command, + org_id.as_deref(), + json, + ) + .await + } ClickPipeCommands::Create { command } => match command { ClickPipeCreateCommands::ObjectStorage(args) => { cloud::commands::clickpipe_create_s3(&client, &args, json).await diff --git a/crates/clickhousectl/tests/cli_request_shape_test.rs b/crates/clickhousectl/tests/cli_request_shape_test.rs index 205e390..7e73c7a 100644 --- a/crates/clickhousectl/tests/cli_request_shape_test.rs +++ b/crates/clickhousectl/tests/cli_request_shape_test.rs @@ -1740,3 +1740,269 @@ async fn agent_session_and_trace_headers_are_forwarded() { traceparent, ); } + +// ── ClickPipe object-storage ingestion-control flags (#289) ───────────────── +// +// `--skip-initial-load` and `--start-after` must serialize to +// `skipInitialLoad` / `startAfter` on the object-storage source body when +// passed, and stay absent when omitted. `--skip-initial-load` requires +// `--queue-url`; `--start-after` conflicts with `--skip-initial-load`. + +#[tokio::test] +async fn s3_skip_initial_load_serializes_when_passed() { + let mock = start_mock_clickpipes_api().await; + let body = invoke_cli_capture_body( + &mock, + &[ + "clickpipe", + "create", + "object-storage", + "svc-id", + "--name", + "t", + "--source-url", + "https://bucket.s3.us-east-1.amazonaws.com/data/*.json", + "--format", + "JSONEachRow", + "--database", + "d", + "--table", + "t", + "--column", + "id:Int64", + "--continuous", + "--queue-url", + "https://sqs.us-east-1.amazonaws.com/123/q", + "--skip-initial-load", + "--org-id", + "org", + ], + ) + .await; + let s3 = &body["source"]["objectStorage"]; + assert_eq!(s3["skipInitialLoad"], true); + assert_eq!(s3["queueUrl"], "https://sqs.us-east-1.amazonaws.com/123/q"); + // startAfter is absent when --start-after not passed. + assert!( + s3.get("startAfter").is_none(), + "startAfter leaked when --start-after not passed: {s3}", + ); +} + +#[tokio::test] +async fn s3_start_after_serializes_when_passed() { + let mock = start_mock_clickpipes_api().await; + let body = invoke_cli_capture_body( + &mock, + &[ + "clickpipe", + "create", + "object-storage", + "svc-id", + "--name", + "t", + "--source-url", + "https://bucket.s3.us-east-1.amazonaws.com/data/*.json", + "--format", + "JSONEachRow", + "--database", + "d", + "--table", + "t", + "--column", + "id:Int64", + "--continuous", + "--queue-url", + "https://sqs.us-east-1.amazonaws.com/123/q", + "--start-after", + "obj-key-001", + "--org-id", + "org", + ], + ) + .await; + let s3 = &body["source"]["objectStorage"]; + assert_eq!(s3["startAfter"], "obj-key-001"); + // skipInitialLoad is absent when --skip-initial-load not passed. + assert!( + s3.get("skipInitialLoad").is_none(), + "skipInitialLoad leaked when --skip-initial-load not passed: {s3}", + ); +} + +// ── ClickPipe MySQL --server-id (#289) ───────────────────────────────────── +// +// `--server-id` must serialize to `serverId` on the MySQL source body when +// passed, and stay absent when omitted. + +#[tokio::test] +async fn mysql_server_id_serializes_when_passed() { + let mock = start_mock_clickpipes_api().await; + let body = invoke_cli_capture_body( + &mock, + &[ + "clickpipe", + "create", + "mysql", + "svc-id", + "--name", + "t", + "--host", + "mysql", + "--port", + "3306", + "--username", + "u", + "--password", + "p", + "--table-mapping", + "mydb.t:t", + "--replication-mode", + "cdc", + "--server-id", + "4242", + "--org-id", + "org", + ], + ) + .await; + let mysql = &body["source"]["mysql"]; + assert_eq!(mysql["serverId"], 4242); +} + +#[tokio::test] +async fn mysql_server_id_absent_when_not_passed() { + let mock = start_mock_clickpipes_api().await; + let body = invoke_cli_capture_body( + &mock, + &[ + "clickpipe", + "create", + "mysql", + "svc-id", + "--name", + "t", + "--host", + "mysql", + "--port", + "3306", + "--username", + "u", + "--password", + "p", + "--table-mapping", + "mydb.t:t", + "--replication-mode", + "cdc", + "--org-id", + "org", + ], + ) + .await; + let mysql = &body["source"]["mysql"]; + assert!( + mysql.get("serverId").is_none(), + "serverId leaked when --server-id not passed: {mysql}", + ); +} + +// ── ClickPipe schema discovery (#289, beta) ──────────────────────────────── +// +// `clickpipe schema-discover` POSTs to .../clickpipes/schemaDiscovery with a +// `source` containing the kafka/kinesis source built from the CLI args. The +// request body shape is asserted; the stubbed response is rendered as a table +// (or JSON with --json). + +/// Start a wiremock server that accepts a schema-discovery POST and records +/// the request body. Returns inferred fields the CLI renders. +async fn start_mock_schema_discovery_api() -> MockServer { + let mock = MockServer::start().await; + let stub_response = serde_json::json!({ + "result": { + "fields": [ + { "name": "id", "type": "Int64", "optional": false }, + { "name": "event", "type": "String", "optional": true }, + ], + }, + "status": 200, + "requestId": "stub-schema-discovery", + }); + Mock::given(method("POST")) + .and(path_regex( + r"^/v1/organizations/[^/]+/services/[^/]+/clickpipes/schemaDiscovery$", + )) + .respond_with(ResponseTemplate::new(200).set_body_json(stub_response)) + .mount(&mock) + .await; + mock +} + +#[tokio::test] +async fn schema_discover_kafka_posts_source_body() { + let mock = start_mock_schema_discovery_api().await; + let body = invoke_cli_capture_body( + &mock, + &[ + "clickpipe", + "schema-discover", + "svc-id", + "--org-id", + "org", + "kafka", + "--brokers", + "broker:9092", + "--topics", + "topic", + "--format", + "JSONEachRow", + "--auth", + "IAM_ROLE", + "--iam-role", + "arn:aws:iam::123:role/x", + ], + ) + .await; + let kafka = &body["source"]["kafka"]; + assert_eq!(kafka["brokers"], "broker:9092"); + assert_eq!(kafka["topics"], "topic"); + assert_eq!(kafka["format"], "JSONEachRow"); + // Kinesis is absent for a Kafka discovery request. + assert!( + body["source"].get("kinesis").is_none(), + "kinesis leaked into kafka schema-discovery body: {}", + body["source"], + ); +} + +#[tokio::test] +async fn schema_discover_kinesis_posts_source_body() { + let mock = start_mock_schema_discovery_api().await; + let body = invoke_cli_capture_body( + &mock, + &[ + "clickpipe", + "schema-discover", + "svc-id", + "--org-id", + "org", + "kinesis", + "--stream-name", + "mystream", + "--region", + "us-east-1", + "--format", + "JSONEachRow", + ], + ) + .await; + let kinesis = &body["source"]["kinesis"]; + assert_eq!(kinesis["streamName"], "mystream"); + assert_eq!(kinesis["region"], "us-east-1"); + assert_eq!(kinesis["format"], "JSONEachRow"); + // Kafka is absent for a Kinesis discovery request. + assert!( + body["source"].get("kafka").is_none(), + "kafka leaked into kinesis schema-discovery body: {}", + body["source"], + ); +}