Connect to a secured ClickHouse catalog: header credentials, verified TLS, bounded retries

docs/integration-api-v1.md

@@ -181,6 +181,31 @@ refuses the persistent backend with `ConfigurationError` rather than generating
 with nothing stored. The catalog
 takes one publisher per `(database, table_prefix)`, so a second engine on the
 same catalog is refused at `create_record_runtime`.
+To reach a secured catalog, set `clickhouse_scheme="https"` (and the server's
+TLS HTTP port, usually 8443) on `NativeCaptureStorageConfig`. The client always
+verifies the server's certificate and name, against libcurl's built-in CA
+bundle/directory, or a private CA given as `clickhouse_ca_file` (a PEM bundle)
+or `clickhouse_ca_path` (a hashed directory). Each replaces libcurl's built-in
+default for that option rather than adding to it: a libcurl built with both a
+bundle and a directory (Debian, Ubuntu) keeps trusting the system roots
+through the other, a bundle-only build (RHEL, Fedora) does not, so to trust
+both there, pass a bundle holding the system roots and the private CA.
+`clickhouse_user`/`clickhouse_password` travel as
+`X-ClickHouse-User`/`X-ClickHouse-Key` headers, never in a URL, and the
+password is left out of the config's repr. A password over plain http is
+refused unless `clickhouse_allow_insecure_http=True`, and `clickhouse_host`
+must be a bare host (no scheme, port or `user:password@`).
+`clickhouse_reader_user`/`clickhouse_reader_password` give
+`NativeCaptureReader` a separate account, for example one limited with
+`GRANT SELECT` on the catalog tables (or a `readonly=2` settings profile). Do
+not use a `readonly=1` profile: the reader sends its query limits
+(`max_rows_to_read`, `max_execution_time`, ...) as settings, which
+`readonly=1` refuses (Code 164, READONLY). The storage service always uses
+`clickhouse_user`. Every catalog request is bounded by
+`clickhouse_request_timeout_s`, which must be at least twice
+`publish_timeout_s` (10 s at the default 5 s publish timeout). A refused connection is retried for any
+statement; a reset, or a 5xx that is not a permanent ClickHouse error (such
+as a row limit or a denied grant), for reads only; and a timeout never.
 The schedule's default factory creates a distinct `CaptureSchedule` for each
 config instance.
 `MonitoringEngine` stores the config, while concrete adaptors decide whether

native/csrc/catalog/bindings_store.cpp

@@ -52,11 +52,27 @@ dmi_store::S3Config s3_config(const py::dict& d) {
   return c;
 }
 
-dc::ClickHouseTimeouts clickhouse_timeouts(const py::dict& d) {
-  dc::ClickHouseTimeouts t;
-  t.connect_s = get<double>(d, "clickhouse_connect_timeout_s", t.connect_s);
-  t.request_s = get<double>(d, "clickhouse_request_timeout_s", t.request_s);
-  return t;
+// The catalog connection, for the service and the reader alike: both read
+// the same keys, so a reader is pointed at a secured catalog exactly as the
+// service is (NativeCaptureReader passes its own account, when configured,
+// under the same key names). The client validates it on construction.
+dc::ClickHouseConnection clickhouse_connection(const py::dict& d) {
+  dc::ClickHouseConnection c;
+  c.scheme = get<std::string>(d, "clickhouse_scheme", c.scheme);
+  c.host = get<std::string>(d, "clickhouse_host", c.host);
+  c.port = get<uint16_t>(d, "clickhouse_port", c.port);
+  c.user = get<std::string>(d, "clickhouse_user", c.user);
+  c.password = get<std::string>(d, "clickhouse_password", c.password);
+  c.ca_file = get<std::string>(d, "clickhouse_ca_file", c.ca_file);
+  c.ca_path = get<std::string>(d, "clickhouse_ca_path", c.ca_path);
+  c.allow_insecure_http =
+      get<bool>(d, "clickhouse_allow_insecure_http", c.allow_insecure_http);
+  c.timeouts.connect_s =
+      get<double>(d, "clickhouse_connect_timeout_s", c.timeouts.connect_s);
+  c.timeouts.request_s =
+      get<double>(d, "clickhouse_request_timeout_s", c.timeouts.request_s);
+  c.max_attempts = get<int>(d, "clickhouse_max_attempts", c.max_attempts);
+  return c;
 }
 
 dc::StorageServiceConfig service_config(const py::dict& d) {
@@ -69,9 +85,7 @@ dc::StorageServiceConfig service_config(const py::dict& d) {
   c.uploader.max_in_flight_bytes =
       get<uint64_t>(d, "uploader_max_in_flight_bytes", c.uploader.max_in_flight_bytes);
   c.uploader.max_attempts = get<int>(d, "uploader_max_attempts", c.uploader.max_attempts);
-  c.clickhouse_host = get<std::string>(d, "clickhouse_host", c.clickhouse_host);
-  c.clickhouse_port = get<uint16_t>(d, "clickhouse_port", c.clickhouse_port);
-  c.clickhouse_timeouts = clickhouse_timeouts(d);
+  c.clickhouse = clickhouse_connection(d);
   c.max_index_attempts = get<int>(d, "max_index_attempts", c.max_index_attempts);
   c.writer.database = get<std::string>(d, "database", "default");
   c.writer.table_prefix = get<std::string>(d, "table_prefix", "dmi");
@@ -184,9 +198,7 @@ class CaptureReader {
   explicit CaptureReader(const py::dict& d)
       : s3_(s3_config(d)),
         client_(std::make_shared<const dc::ClickHouseClient>(
-            get<std::string>(d, "clickhouse_host", "127.0.0.1"),
-            get<uint16_t>(d, "clickhouse_port", 8123),
-            clickhouse_timeouts(d))),
+            clickhouse_connection(d))),
         config_{get<std::string>(d, "database", "default"),
                 get<std::string>(d, "table_prefix", "dmi")},
         catalog_(client_, config_),