diff --git a/README.md b/README.md index 7202b2c5..cfb2fffe 100644 --- a/README.md +++ b/README.md @@ -770,14 +770,26 @@ clickhousectl cloud clickpipe create kinesis \ --database default --table events \ --column "event_id:Int64" --column "name:String" -# From PostgreSQL (CDC) +# From PostgreSQL (CDC) with a publicly trusted TLS certificate +# TLS and certificate verification are enabled by default. The certificate +# hostname defaults to --host. clickhousectl cloud clickpipe create postgres \ --name my-pg-pipe \ --host db.example.com --pg-database mydb \ --username "$POSTGRES_USERNAME" --password "$POSTGRES_PASSWORD" \ + --publication-name clickpipes \ --table-mapping "public.users:public_users" \ --table-mapping "public.orders:public_orders" +# From PostgreSQL with a private or self-signed certificate +clickhousectl cloud clickpipe create postgres \ + --name my-private-pg-pipe \ + --host db.private.example.com --pg-database mydb \ + --username "$POSTGRES_USERNAME" --password "$POSTGRES_PASSWORD" \ + --publication-name clickpipes \ + --ca-certificate ./postgres-ca.pem \ + --table-mapping "public.users:public_users" + # 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) @@ -803,6 +815,15 @@ clickhousectl cloud clickpipe create bigquery \ --table-mapping "dataset.table:target_table" ``` +Before creating a PostgreSQL CDC ClickPipe: + +- Make the source reachable from ClickPipes and allow the [ClickPipes static IPs](https://clickhouse.com/docs/integrations/clickpipes/networking/static-ips) for your service region. +- Enable logical replication on PostgreSQL. +- Create a publication named `clickpipes` that includes every source table passed with `--table-mapping`. +- Grant the ClickPipes user schema `USAGE`, table `SELECT`, and replication privileges. + +See the [PostgreSQL ClickPipes setup guide](https://clickhouse.com/docs/integrations/clickpipes/postgres) for provider-specific prerequisites or the [generic PostgreSQL source setup](https://clickhouse.com/docs/integrations/clickpipes/postgres/source/generic) for self-hosted and other providers. `--ca-certificate` is needed only when the source certificate is signed by a private CA or is self-signed. Use `--tls-host` when the hostname in that certificate differs from `--host`. + Use `clickhousectl cloud clickpipe create --help` for the full list of options per source type. #### Discovering a source schema (beta) diff --git a/crates/clickhousectl/src/cloud/clickpipes.rs b/crates/clickhousectl/src/cloud/clickpipes.rs index 5d885b2c..8b155b38 100644 --- a/crates/clickhousectl/src/cloud/clickpipes.rs +++ b/crates/clickhousectl/src/cloud/clickpipes.rs @@ -328,6 +328,10 @@ pub enum ClickPipeCreateCommands { Kinesis(KinesisCreateArgs), /// Create a ClickPipe from PostgreSQL + #[command( + long_about = "Create a ClickPipe from PostgreSQL.\n\nTLS and certificate verification are enabled by default. Omit --ca-certificate when the source certificate has a publicly trusted chain. For a private or self-signed source certificate, pass its PEM CA bundle with --ca-certificate. Certificate hostname verification uses --host unless --tls-host is provided.", + after_long_help = "PostgreSQL ClickPipes setup:\n https://clickhouse.com/docs/integrations/clickpipes/postgres\nGeneric PostgreSQL source setup:\n https://clickhouse.com/docs/integrations/clickpipes/postgres/source/generic\nClickPipes networking and static IPs:\n https://clickhouse.com/docs/integrations/clickpipes/networking/static-ips" + )] Postgres(PostgresCreateArgs), /// Create a ClickPipe from MySQL @@ -718,11 +722,11 @@ pub struct PostgresCreateArgs { #[arg(long, required_if_eq("auth", "IAM_ROLE"))] pub iam_role: Option, - /// TLS hostname + /// Certificate hostname to verify (defaults to --host) #[arg(long)] pub tls_host: Option, - /// Path to CA certificate file + /// Path to a PEM CA bundle for a private or self-signed source certificate #[arg(long)] pub ca_certificate: Option, @@ -1805,11 +1809,35 @@ async fn clickpipe_create_postgres( let org_id = resolve_org_id(client, args.org_id.as_deref()).await?; let clickpipe = client .create_clickpipe(&org_id, &args.service_id, &request) - .await?; + .await + .map_err(add_postgres_tls_guidance)?; print_created(&clickpipe, json)?; Ok(()) } +fn add_postgres_tls_guidance(mut error: CloudError) -> CloudError { + let message = error.message.to_ascii_lowercase(); + let hint = if message.contains("x509: certificate signed by unknown authority") { + Some("Provide the PostgreSQL source's PEM CA bundle with --ca-certificate .") + } else if message.contains("x509: hostname mismatch") + || (message.contains("x509: certificate is valid for ") && message.contains(", not ")) + || (message.contains("x509: cannot validate certificate for ") + && message.contains(" because it doesn't contain any ip sans")) + { + Some( + "Set --tls-host to a DNS name covered by the PostgreSQL source certificate (it defaults to --host).", + ) + } else { + None + }; + + if let Some(hint) = hint { + error.message.push_str("\n\nHint: "); + error.message.push_str(hint); + } + error +} + fn build_postgres_request( args: &PostgresCreateArgs, ) -> CloudResult { @@ -2322,6 +2350,45 @@ mod tests { ); } + #[test] + fn postgres_create_help_explains_tls_defaults_and_prerequisites() { + let error = Cli::try_parse_from([ + "clickhousectl", + "cloud", + "clickpipe", + "create", + "postgres", + "--help", + ]) + .err() + .expect("--help should stop parsing"); + assert_eq!(error.kind(), clap::error::ErrorKind::DisplayHelp); + + let help = error.to_string(); + for expected in [ + "TLS and certificate verification are enabled by default", + "Omit --ca-certificate when the source certificate has a publicly trusted chain", + "private or self-signed source certificate", + "Certificate hostname to verify (defaults to --host)", + "https://clickhouse.com/docs/integrations/clickpipes/postgres", + "https://clickhouse.com/docs/integrations/clickpipes/postgres/source/generic", + "https://clickhouse.com/docs/integrations/clickpipes/networking/static-ips", + ] { + assert!( + help.contains(expected), + "missing {expected:?} in help:\n{help}" + ); + } + assert!( + !help.contains("--disable-tls"), + "unexpected TLS bypass flag" + ); + assert!( + !help.contains("--skip-cert-verification"), + "unexpected certificate bypass flag" + ); + } + fn assert_object_storage_value(flag: &str, value: &str) { if flag == "--format" { parse_clickpipe(&[ diff --git a/crates/clickhousectl/tests/cli_request_shape_test.rs b/crates/clickhousectl/tests/cli_request_shape_test.rs index dd43d1f0..4f43ff11 100644 --- a/crates/clickhousectl/tests/cli_request_shape_test.rs +++ b/crates/clickhousectl/tests/cli_request_shape_test.rs @@ -2350,6 +2350,63 @@ fn postgres_args_minimal() -> Vec { .collect() } +async fn invoke_postgres_create_api_error(message: &str) -> std::process::Output { + let mock = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/v1/organizations/org/services/svc-id/clickpipes")) + .respond_with(ResponseTemplate::new(400).set_body_json(serde_json::json!({ + "status": 400, + "error": message, + "requestId": "stub-postgres-create-error", + }))) + .mount(&mock) + .await; + + let args = postgres_args_minimal(); + let arg_refs: Vec<&str> = args.iter().map(String::as_str).collect(); + invoke_cli_with_cloud_credentials(&mock, &arg_refs) +} + +#[tokio::test] +async fn postgres_unknown_authority_error_preserves_detail_and_names_ca_flag() { + let api_error = "BAD_REQUEST: failed to establish connection: tls: failed to verify certificate: x509: certificate signed by unknown authority"; + let output = invoke_postgres_create_api_error(api_error).await; + + assert_eq!(output.status.code(), Some(1)); + assert_eq!( + String::from_utf8_lossy(&output.stderr), + format!( + "Error: {api_error}\n\nHint: Provide the PostgreSQL source's PEM CA bundle with --ca-certificate .\n" + ) + ); +} + +#[tokio::test] +async fn postgres_hostname_mismatch_error_preserves_detail_and_names_tls_host_flag() { + let api_error = "BAD_REQUEST: failed to establish connection: tls: failed to verify certificate: x509: certificate is valid for pg.example.com, not db.example.com"; + let output = invoke_postgres_create_api_error(api_error).await; + + assert_eq!(output.status.code(), Some(1)); + assert_eq!( + String::from_utf8_lossy(&output.stderr), + format!( + "Error: {api_error}\n\nHint: Set --tls-host to a DNS name covered by the PostgreSQL source certificate (it defaults to --host).\n" + ) + ); +} + +#[tokio::test] +async fn postgres_unrelated_connection_error_has_no_tls_hint() { + let api_error = "BAD_REQUEST: failed to establish connection: connection refused"; + let output = invoke_postgres_create_api_error(api_error).await; + + assert_eq!(output.status.code(), Some(1)); + assert_eq!( + String::from_utf8_lossy(&output.stderr), + format!("Error: {api_error}\n") + ); +} + #[tokio::test] async fn postgres_invalid_inputs_fail_before_cloud_api_dispatch() { let without_org = || {