#include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include "config.h" #include #if defined(OS_LINUX) # include # include # include # include #endif #if USE_SSL # include # include # include #endif #if USE_SSH && defined(OS_LINUX) # include # include # include #endif #if USE_GRPC # include #endif #if USE_NURAFT # include # include # include #endif #if USE_AZURE_BLOB_STORAGE # include # include #endif /// A minimal file used when the server is run without installation constexpr unsigned char resource_embedded_xml[] = { #embed "embedded.xml" }; namespace DB { namespace Setting { extern const SettingsSeconds http_receive_timeout; extern const SettingsSeconds http_send_timeout; extern const SettingsSeconds receive_timeout; extern const SettingsSeconds send_timeout; } namespace ServerSetting { extern const ServerSettingsUInt32 allow_feature_tier; extern const ServerSettingsUInt32 asynchronous_heavy_metrics_update_period_s; extern const ServerSettingsUInt32 asynchronous_metrics_update_period_s; extern const ServerSettingsBool asynchronous_metrics_enable_heavy_metrics; extern const ServerSettingsBool asynchronous_metrics_keeper_metrics_only; extern const ServerSettingsBool async_insert_queue_flush_on_shutdown; extern const ServerSettingsUInt64 async_insert_threads; extern const ServerSettingsBool async_load_databases; extern const ServerSettingsBool async_load_system_database; extern const ServerSettingsUInt64 background_buffer_flush_schedule_pool_size; extern const ServerSettingsUInt64 background_common_pool_size; extern const ServerSettingsUInt64 background_distributed_schedule_pool_size; extern const ServerSettingsUInt64 background_fetches_pool_size; extern const ServerSettingsFloat background_merges_mutations_concurrency_ratio; extern const ServerSettingsString background_merges_mutations_scheduling_policy; extern const ServerSettingsUInt64 background_message_broker_schedule_pool_size; extern const ServerSettingsUInt64 background_move_pool_size; extern const ServerSettingsUInt64 background_pool_size; extern const ServerSettingsUInt64 background_schedule_pool_size; extern const ServerSettingsUInt64 background_streaming_schedule_pool_size; extern const ServerSettingsUInt64 backups_io_thread_pool_queue_size; extern const ServerSettingsDouble cache_size_to_ram_max_ratio; extern const ServerSettingsString columns_cache_policy; extern const ServerSettingsUInt64 columns_cache_size; extern const ServerSettingsDouble columns_cache_size_ratio; extern const ServerSettingsDouble columns_cache_size_to_ram_ratio; extern const ServerSettingsDouble columns_cache_free_memory_ratio; extern const ServerSettingsUInt64 columns_cache_history_window_ms; extern const ServerSettingsDouble cannot_allocate_thread_fault_injection_probability; extern const ServerSettingsUInt64 cgroups_memory_usage_observer_wait_time; extern const ServerSettingsUInt64 compiled_expression_cache_elements_size; extern const ServerSettingsUInt64 compiled_expression_cache_size; extern const ServerSettingsUInt64 concurrent_threads_soft_limit_num; extern const ServerSettingsUInt64 concurrent_threads_soft_limit_ratio_to_cores; extern const ServerSettingsString concurrent_threads_scheduler; extern const ServerSettingsBool concurrent_threads_lazy_allocation; extern const ServerSettingsUInt64 config_reload_interval_ms; extern const ServerSettingsUInt64 database_catalog_drop_table_concurrency; extern const ServerSettingsUInt64 database_catalog_shutdown_table_concurrency; extern const ServerSettingsString default_database; extern const ServerSettingsString insert_deduplication_version; extern const ServerSettingsBool disable_internal_dns_cache; extern const ServerSettingsBool s3queue_disable_streaming; extern const ServerSettingsBool enable_read_through_distributed_cache; extern const ServerSettingsBool enable_write_through_distributed_cache; extern const ServerSettingsUInt64 disk_connections_soft_limit; extern const ServerSettingsUInt64 disk_connections_store_limit; extern const ServerSettingsUInt64 disk_connections_hard_limit; extern const ServerSettingsUInt64 disk_connections_warn_limit; extern const ServerSettingsUInt64 disk_connections_rcvbuf; extern const ServerSettingsUInt64 disk_connections_sndbuf; extern const ServerSettingsBool dns_allow_resolve_names_to_ipv4; extern const ServerSettingsBool dns_allow_resolve_names_to_ipv6; extern const ServerSettingsUInt64 dns_cache_max_entries; extern const ServerSettingsInt32 dns_cache_update_period; extern const ServerSettingsUInt32 dns_max_consecutive_failures; extern const ServerSettingsBool enable_azure_sdk_logging; extern const ServerSettingsBool enable_silk_runtime; extern const ServerSettingsBool enable_gpu; extern const ServerSettingsUInt64 global_profiler_cpu_time_period_ns; extern const ServerSettingsUInt64 global_profiler_real_time_period_ns; extern const ServerSettingsUInt64 http_connections_soft_limit; extern const ServerSettingsUInt64 http_connections_store_limit; extern const ServerSettingsUInt64 http_connections_hard_limit; extern const ServerSettingsUInt64 http_connections_warn_limit; extern const ServerSettingsUInt64 http_connections_rcvbuf; extern const ServerSettingsUInt64 http_connections_sndbuf; extern const ServerSettingsString index_mark_cache_policy; extern const ServerSettingsUInt64 index_mark_cache_size; extern const ServerSettingsDouble index_mark_cache_size_ratio; extern const ServerSettingsString vector_similarity_index_cache_policy; extern const ServerSettingsUInt64 vector_similarity_index_cache_size; extern const ServerSettingsUInt64 vector_similarity_index_cache_max_entries; extern const ServerSettingsDouble vector_similarity_index_cache_size_ratio; extern const ServerSettingsString text_index_tokens_cache_policy; extern const ServerSettingsUInt64 text_index_tokens_cache_size; extern const ServerSettingsUInt64 text_index_tokens_cache_max_entries; extern const ServerSettingsDouble text_index_tokens_cache_size_ratio; extern const ServerSettingsString text_index_header_cache_policy; extern const ServerSettingsUInt64 text_index_header_cache_size; extern const ServerSettingsUInt64 text_index_header_cache_max_entries; extern const ServerSettingsDouble text_index_header_cache_size_ratio; extern const ServerSettingsString text_index_postings_cache_policy; extern const ServerSettingsUInt64 text_index_postings_cache_size; extern const ServerSettingsUInt64 text_index_postings_cache_max_entries; extern const ServerSettingsDouble text_index_postings_cache_size_ratio; extern const ServerSettingsString index_uncompressed_cache_policy; extern const ServerSettingsUInt64 index_uncompressed_cache_size; extern const ServerSettingsDouble index_uncompressed_cache_size_ratio; extern const ServerSettingsString iceberg_metadata_files_cache_policy; extern const ServerSettingsUInt64 iceberg_metadata_files_cache_size; extern const ServerSettingsUInt64 iceberg_metadata_files_cache_max_entries; extern const ServerSettingsDouble iceberg_metadata_files_cache_size_ratio; extern const ServerSettingsString paimon_metadata_files_cache_policy; extern const ServerSettingsUInt64 paimon_metadata_files_cache_size; extern const ServerSettingsUInt64 paimon_metadata_files_cache_max_entries; extern const ServerSettingsDouble paimon_metadata_files_cache_size_ratio; extern const ServerSettingsString parquet_metadata_cache_policy; extern const ServerSettingsUInt64 parquet_metadata_cache_size; extern const ServerSettingsUInt64 parquet_metadata_cache_max_entries; extern const ServerSettingsDouble parquet_metadata_cache_size_ratio; extern const ServerSettingsUInt64 io_thread_pool_queue_size; extern const ServerSettingsBool jemalloc_enable_global_profiler; extern const ServerSettingsBool jemalloc_collect_global_profile_samples_in_trace_log; extern const ServerSettingsBool jemalloc_enable_background_threads; extern const ServerSettingsUInt64 jemalloc_max_background_threads_num; extern const ServerSettingsUInt64 jemalloc_profiler_sampling_rate; extern const ServerSettingsSeconds keep_alive_timeout; extern const ServerSettingsString mark_cache_policy; extern const ServerSettingsUInt64 mark_cache_size; extern const ServerSettingsDouble mark_cache_size_ratio; extern const ServerSettingsString unique_key_index_cache_policy; extern const ServerSettingsUInt64 unique_key_index_cache_size_bytes; extern const ServerSettingsDouble unique_key_index_cache_size_ratio; extern const ServerSettingsString unique_key_bitmap_cache_policy; extern const ServerSettingsUInt64 unique_key_bitmap_cache_size_bytes; extern const ServerSettingsDouble unique_key_bitmap_cache_size_ratio; extern const ServerSettingsUInt64 max_fetch_partition_thread_pool_size; extern const ServerSettingsUInt64 max_active_parts_loading_thread_pool_size; extern const ServerSettingsUInt64 max_backups_io_thread_pool_free_size; extern const ServerSettingsUInt64 max_backups_io_thread_pool_size; extern const ServerSettingsUInt64 max_concurrent_insert_queries; extern const ServerSettingsUInt64 max_concurrent_queries; extern const ServerSettingsUInt64 max_concurrent_select_queries; extern const ServerSettingsInt32 max_connections; extern const ServerSettingsUInt64 max_database_num_to_warn; extern const ServerSettingsUInt32 max_database_replicated_create_table_thread_pool_size; extern const ServerSettingsUInt64 max_dictionary_num_to_warn; extern const ServerSettingsUInt64 max_io_thread_pool_free_size; extern const ServerSettingsUInt64 max_io_thread_pool_size; extern const ServerSettingsUInt64 max_keep_alive_requests; extern const ServerSettingsUInt64 max_outdated_parts_loading_thread_pool_size; extern const ServerSettingsUInt64 max_per_cpu_untracked_memory; extern const ServerSettingsUInt64 max_partition_size_to_drop; extern const ServerSettingsUInt64 max_part_num_to_warn; extern const ServerSettingsUInt64 max_pending_mutations_to_warn; extern const ServerSettingsUInt64 max_pending_mutations_execution_time_to_warn; extern const ServerSettingsUInt64 max_parts_cleaning_thread_pool_size; extern const ServerSettingsUInt64 max_named_collection_num_to_warn; extern const ServerSettingsUInt64 max_named_collection_num_to_throw; extern const ServerSettingsUInt64 max_table_num_to_throw; extern const ServerSettingsUInt64 max_replicated_table_num_to_throw; extern const ServerSettingsUInt64 max_view_num_to_throw; extern const ServerSettingsUInt64 max_dictionary_num_to_throw; extern const ServerSettingsUInt64 max_database_num_to_throw; extern const ServerSettingsUInt64 max_remote_read_network_bandwidth_for_server; extern const ServerSettingsUInt64 max_remote_write_network_bandwidth_for_server; extern const ServerSettingsUInt64 max_remote_read_connections; extern const ServerSettingsUInt64 reader_executor_memory_pressure_elevated_level_pct; extern const ServerSettingsUInt64 reader_executor_memory_pressure_high_level_pct; extern const ServerSettingsUInt64 reader_executor_memory_pressure_critical_level_pct; extern const ServerSettingsUInt64 max_local_read_bandwidth_for_server; extern const ServerSettingsUInt64 max_local_write_bandwidth_for_server; extern const ServerSettingsUInt64 max_server_memory_usage; extern const ServerSettingsDouble max_server_memory_usage_to_ram_ratio; extern const ServerSettingsUInt64 max_table_num_to_warn; extern const ServerSettingsUInt64 max_table_size_to_drop; extern const ServerSettingsUInt64 max_temporary_data_on_disk_size; extern const ServerSettingsUInt64 max_thread_pool_free_size; extern const ServerSettingsUInt64 max_thread_pool_size; extern const ServerSettingsUInt64 max_unexpected_parts_loading_thread_pool_size; extern const ServerSettingsUInt64 max_view_num_to_warn; extern const ServerSettingsUInt64 max_waiting_queries; extern const ServerSettingsUInt64 memory_worker_period_ms; extern const ServerSettingsDouble memory_worker_purge_dirty_pages_threshold_ratio; extern const ServerSettingsDouble memory_worker_purge_total_memory_threshold_ratio; extern const ServerSettingsUInt64 memory_worker_decay_adjustment_period_ms; extern const ServerSettingsBool memory_worker_correct_memory_tracker; extern const ServerSettingsBool memory_worker_use_cgroup; extern const ServerSettingsDouble memory_worker_rss_speculative_reserve_ratio; extern const ServerSettingsBool memory_worker_dynamic_hard_limit; extern const ServerSettingsUInt64 merges_mutations_memory_usage_soft_limit; extern const ServerSettingsDouble merges_mutations_memory_usage_to_ram_ratio; extern const ServerSettingsString merge_workload; extern const ServerSettingsUInt64 min_allocation_size_to_log_stack_trace; extern const ServerSettingsUInt64 min_allocation_size_to_throw_on_memory_limit; extern const ServerSettingsUInt64 mmap_cache_size; extern const ServerSettingsString mutation_workload; extern const ServerSettingsString query_condition_cache_policy; extern const ServerSettingsUInt64 query_condition_cache_size; extern const ServerSettingsDouble query_condition_cache_size_ratio; extern const ServerSettingsString encryption_header_cache_policy; extern const ServerSettingsUInt64 encryption_header_cache_size; extern const ServerSettingsDouble encryption_header_cache_size_ratio; extern const ServerSettingsBool prepare_system_log_tables_on_startup; extern const ServerSettingsBool user_profile_events_per_cpu; extern const ServerSettingsBool show_addresses_in_stack_traces; extern const ServerSettingsBool shutdown_wait_backups_and_restores; extern const ServerSettingsUInt64 shutdown_wait_unfinished; extern const ServerSettingsBool shutdown_wait_unfinished_queries; extern const ServerSettingsUInt64 storage_connections_soft_limit; extern const ServerSettingsUInt64 storage_connections_store_limit; extern const ServerSettingsUInt64 storage_connections_hard_limit; extern const ServerSettingsUInt64 storage_connections_warn_limit; extern const ServerSettingsUInt64 storage_connections_rcvbuf; extern const ServerSettingsUInt64 storage_connections_sndbuf; extern const ServerSettingsUInt64 tables_loader_background_pool_size; extern const ServerSettingsUInt64 tables_loader_foreground_pool_size; extern const ServerSettingsString temporary_data_in_cache; extern const ServerSettingsUInt64 thread_pool_queue_size; extern const ServerSettingsString tmp_policy; extern const ServerSettingsUInt64 total_memory_profiler_sample_max_allocation_size; extern const ServerSettingsUInt64 total_memory_profiler_sample_min_allocation_size; extern const ServerSettingsUInt64 total_memory_profiler_step; extern const ServerSettingsDouble total_memory_tracker_sample_probability; extern const ServerSettingsBool throw_on_unknown_workload; extern const ServerSettingsBool cpu_slot_preemption; extern const ServerSettingsUInt64 cpu_slot_quantum_ns; extern const ServerSettingsUInt64 cpu_slot_preemption_timeout_ms; extern const ServerSettingsBool cpu_slot_parking; extern const ServerSettingsString uncompressed_cache_policy; extern const ServerSettingsUInt64 uncompressed_cache_size; extern const ServerSettingsDouble uncompressed_cache_size_ratio; extern const ServerSettingsUInt64 per_cpu_untracked_memory_thread_buffer; extern const ServerSettingsBool use_separate_cache_arena; extern const ServerSettingsString primary_index_cache_policy; extern const ServerSettingsUInt64 primary_index_cache_size; extern const ServerSettingsDouble primary_index_cache_size_ratio; extern const ServerSettingsString statistics_cache_policy; extern const ServerSettingsUInt64 statistics_cache_size; extern const ServerSettingsDouble statistics_cache_size_ratio; extern const ServerSettingsUInt64 point_in_polygon_cache_size; extern const ServerSettingsBool dictionaries_lazy_load; extern const ServerSettingsBool wait_dictionaries_load_at_startup; extern const ServerSettingsUInt64 max_prefixes_deserialization_thread_pool_size; extern const ServerSettingsUInt64 max_prefixes_deserialization_thread_pool_free_size; extern const ServerSettingsUInt64 prefixes_deserialization_thread_pool_thread_pool_queue_size; extern const ServerSettingsUInt64 max_format_parsing_thread_pool_size; extern const ServerSettingsUInt64 max_format_parsing_thread_pool_free_size; extern const ServerSettingsUInt64 format_parsing_thread_pool_queue_size; extern const ServerSettingsNonZeroUInt64 ai_function_max_thread_pool_size; extern const ServerSettingsUInt64 ai_function_max_thread_pool_free_size; extern const ServerSettingsUInt64 ai_function_thread_pool_queue_size; extern const ServerSettingsUInt64 page_cache_history_window_ms; extern const ServerSettingsString page_cache_policy; extern const ServerSettingsDouble page_cache_size_ratio; extern const ServerSettingsUInt64 page_cache_min_size; extern const ServerSettingsUInt64 page_cache_max_size; extern const ServerSettingsDouble page_cache_free_memory_ratio; extern const ServerSettingsUInt64 page_cache_shards; extern const ServerSettingsUInt64 os_cpu_busy_time_threshold; extern const ServerSettingsFloat min_os_cpu_wait_time_ratio_to_drop_connection; extern const ServerSettingsFloat max_os_cpu_wait_time_ratio_to_drop_connection; extern const ServerSettingsBool skip_binary_checksum_checks; extern const ServerSettingsBool abort_on_logical_error; extern const ServerSettingsBool collect_connection_metrics; extern const ServerSettingsUInt64 jemalloc_flush_profile_interval_bytes; extern const ServerSettingsUInt64 jemalloc_merge_tree_arenas; extern const ServerSettingsBool jemalloc_flush_profile_on_memory_exceeded; extern const ServerSettingsUInt64 jemalloc_flush_profile_on_memory_exceeded_interval; extern const ServerSettingsString allowed_disks_for_table_engines; extern const ServerSettingsUInt64 s3_credentials_provider_max_cache_size; extern const ServerSettingsUInt64 max_open_files; extern const ServerSettingsString path; extern const ServerSettingsString user_files_path; extern const ServerSettingsString user_scripts_path; extern const ServerSettingsString dynamic_user_defined_executable_functions_path; extern const ServerSettingsString top_level_domains_path; extern const ServerSettingsString interserver_http_host; extern const ServerSettingsUInt64 interserver_http_port; extern const ServerSettingsString interserver_https_host; extern const ServerSettingsUInt64 interserver_https_port; extern const ServerSettingsString include_from; extern const ServerSettingsString tmp_path; extern const ServerSettingsString format_schema_path; extern const ServerSettingsString google_protos_path; extern const ServerSettingsString filesystem_caches_path; extern const ServerSettingsInt32 oom_score; extern const ServerSettingsBool oom_canary_enable; extern const ServerSettingsUInt64 oom_canary_size; extern const ServerSettingsBool oom_canary_relaunch; extern const ServerSettingsUInt64 oom_canary_max_rapid_relaunches; extern const ServerSettingsUInt64 oom_canary_initial_backoff_seconds; extern const ServerSettingsUInt64 oom_canary_max_backoff_seconds; extern const ServerSettingsBool remap_executable; extern const ServerSettingsBool mlock_executable; extern const ServerSettingsUInt64 mlock_executable_min_total_memory_amount_bytes; extern const ServerSettingsSeccompMode seccomp; extern const ServerSettingsUInt32 listen_backlog; extern const ServerSettingsBool listen_reuse_port; extern const ServerSettingsBool listen_try; extern const ServerSettingsBool mysql_require_secure_transport; extern const ServerSettingsBool postgresql_require_secure_transport; extern const ServerSettingsBool skip_check_for_incorrect_settings; extern const ServerSettingsUInt64 query_cache_max_size_in_bytes; extern const ServerSettingsUInt64 query_cache_max_entries; extern const ServerSettingsUInt64 query_cache_max_entry_size_in_bytes; extern const ServerSettingsUInt64 query_cache_max_entry_size_in_rows; extern const ServerSettingsString logger_log; extern const ServerSettingsString logger_level; extern const ServerSettingsString logger_startup_level; extern const ServerSettingsString logger_shutdown_level; extern const ServerSettingsString openssl_server_certificate_file; extern const ServerSettingsString openssl_server_private_key_file; extern const ServerSettingsString openssl_server_ca_config; extern const ServerSettingsString openssl_client_certificate_file; extern const ServerSettingsString openssl_client_private_key_file; extern const ServerSettingsString openssl_client_ca_config; extern const ServerSettingsString distributed_ddl_path; extern const ServerSettingsString distributed_ddl_replicas_path; extern const ServerSettingsInt32 distributed_ddl_pool_size; extern const ServerSettingsBool prometheus_keeper_metrics_only; extern const ServerSettingsBool startup_scripts_throw_on_error; extern const ServerSettingsUInt64 keeper_server_socket_receive_timeout_sec; extern const ServerSettingsUInt64 keeper_server_socket_send_timeout_sec; extern const ServerSettingsString hdfs_libhdfs3_conf; extern const ServerSettingsString config_file; extern const ServerSettingsString users_to_ignore_early_memory_limit_check; } namespace ErrorCodes { extern const int STARTUP_SCRIPTS_ERROR; } namespace FileCacheSetting { extern const FileCacheSettingsBool load_metadata_asynchronously; } } namespace CurrentMetrics { extern const Metric Revision; extern const Metric VersionInteger; extern const Metric MemoryTracking; extern const Metric MergesMutationsMemoryTracking; extern const Metric MaxDDLEntryID; extern const Metric MaxPushedDDLEntryID; extern const Metric StartupScriptsExecutionState; extern const Metric IsServerShuttingDown; } namespace DimensionalMetrics { extern MetricFamily & StartupScriptsFailureReason; } namespace ProfileEvents { extern const Event MainConfigLoads; extern const Event ServerStartupMilliseconds; extern const Event InterfaceNativeSendBytes; extern const Event InterfaceNativeReceiveBytes; extern const Event InterfaceHTTPSendBytes; extern const Event InterfaceHTTPReceiveBytes; extern const Event InterfacePrometheusSendBytes; extern const Event InterfacePrometheusReceiveBytes; extern const Event InterfaceInterserverSendBytes; extern const Event InterfaceInterserverReceiveBytes; extern const Event InterfaceMySQLSendBytes; extern const Event InterfaceMySQLReceiveBytes; extern const Event InterfacePostgreSQLSendBytes; extern const Event InterfacePostgreSQLReceiveBytes; } namespace fs = std::filesystem; int mainEntryClickHouseServer(int argc, char ** argv); int mainEntryClickHouseServer(int argc, char ** argv) { DB::Server app; /// Do not fork separate process from watchdog if we attached to terminal. /// Otherwise it breaks gdb usage. /// Can be overridden by environment variable (cannot use server config at this moment). if (argc > 0) { const char * env_watchdog = getenv("CLICKHOUSE_WATCHDOG_ENABLE"); // NOLINT(concurrency-mt-unsafe) if (env_watchdog) { if (0 == strcmp(env_watchdog, "1")) app.shouldSetupWatchdog(argv[0]); /// Other values disable watchdog explicitly. } else if (!isatty(STDIN_FILENO) && !isatty(STDOUT_FILENO) && !isatty(STDERR_FILENO)) app.shouldSetupWatchdog(argv[0]); } try { return app.run(argc, argv); } catch (...) { std::cerr << DB::getCurrentExceptionMessage(true) << "\n"; auto code = DB::getCurrentExceptionCode(); return static_cast(code) ? code : 1; } } namespace DB { namespace ErrorCodes { extern const int ABORTED; extern const int NO_ELEMENTS_IN_CONFIG; extern const int SUPPORT_IS_DISABLED; extern const int ARGUMENT_OUT_OF_BOUND; extern const int EXCESSIVE_ELEMENT_IN_CONFIG; extern const int INVALID_CONFIG_PARAMETER; extern const int INVALID_SETTING_VALUE; extern const int NETWORK_ERROR; extern const int CORRUPTED_DATA; extern const int BAD_ARGUMENTS; } enum StartupScriptsExecutionState : CurrentMetrics::Value { NotFinished = 0, Success = 1, Failure = 2, }; namespace { std::string getCanonicalPath(std::string && path, const std::string & base = {}) { Poco::trimInPlace(path); if (path.empty()) throw Exception(ErrorCodes::INVALID_CONFIG_PARAMETER, "path configuration parameter is empty"); if (!base.empty() && fs::path(path).is_relative()) path = fs::weakly_canonical(fs::path(base) / path); if (path.back() != '/') path += '/'; return std::move(path); } Poco::Net::TCPServerParams::Ptr makeServerParams(const ServerSettings & server_settings) { Poco::Net::TCPServerParams::Ptr params = new Poco::Net::TCPServerParams(); params->setMaxQueued(server_settings[ServerSetting::listen_backlog]); return params; } #if defined(OS_LINUX) /// Whether a debugger is attached to this process, according to the `TracerPid` field of /// `/proc/self/status`, which the kernel sets to the pid of the tracer and to zero when there is /// none. This is a plain read, so unlike the `ptrace(PTRACE_TRACEME)` probe it used to be, it /// answers the same way whatever the `seccomp` server setting denies. bool isRunUnderDebugger() { try { ReadBufferFromFile status("/proc/self/status"); while (!status.eof()) { String line; readStringUntilNewlineInto(line, status); if (!status.eof()) ++status.position(); static constexpr std::string_view prefix = "TracerPid:"; if (!line.starts_with(prefix)) continue; UInt64 tracer_pid = 0; ReadBufferFromString value(line); value.ignore(prefix.size()); skipWhitespaceIfAny(value); readIntText(tracer_pid, value); return tracer_pid != 0; } } catch (...) { tryLogCurrentException(__PRETTY_FUNCTION__); } return false; } #endif } Poco::Net::SocketAddress Server::socketBindListen( const ServerSettings & server_settings, Poco::Net::ServerSocket & socket, const std::string & host, UInt16 port, [[maybe_unused]] bool secure) const { return DB::socketBindListen(server_settings, socket, host, port, &logger()); } namespace { Strings getListenHosts(const Poco::Util::AbstractConfiguration & config) { auto listen_hosts = DB::getMultipleValuesFromConfig(config, "", "listen_host"); if (listen_hosts.empty()) { listen_hosts.emplace_back("::1"); listen_hosts.emplace_back("127.0.0.1"); } return listen_hosts; } Strings getInterserverListenHosts(const Poco::Util::AbstractConfiguration & config) { auto interserver_listen_hosts = DB::getMultipleValuesFromConfig(config, "", "interserver_listen_host"); if (!interserver_listen_hosts.empty()) return interserver_listen_hosts; /// Use more general restriction in case of emptiness return getListenHosts(config); } bool isIntrospectionProtocol(const Poco::Util::AbstractConfiguration & config, const std::string & protocol) { return config.getBool("protocols." + protocol + ".introspection", false); } bool hasIntrospectionProtocols(const Poco::Util::AbstractConfiguration & config) { Poco::Util::AbstractConfiguration::Keys protocols; config.keys("protocols", protocols); return std::any_of(protocols.begin(), protocols.end(), [&](const auto & protocol) { return isIntrospectionProtocol(config, protocol); }); } bool getListenTry(const Poco::Util::AbstractConfiguration & config, const ServerSettings & server_settings) { bool listen_try = server_settings[ServerSetting::listen_try]; if (!listen_try) { Poco::Util::AbstractConfiguration::Keys protocols; config.keys("protocols", protocols); listen_try = DB::getMultipleValuesFromConfig(config, "", "listen_host").empty() && std::none_of(protocols.begin(), protocols.end(), [&](const auto & protocol) { return config.has("protocols." + protocol + ".host") && config.has("protocols." + protocol + ".port"); }); } return listen_try; } #if USE_GRPC || USE_ARROWFLIGHT bool isWildcardListenHost(const std::string & listen_host) { Poco::Net::IPAddress address; return Poco::Net::IPAddress::tryParse(listen_host, address) && address.isWildcard(); } /// gRPC treats `::` and `0.0.0.0` as one family-agnostic wildcard: for either of them it binds a /// dual-stack `::` socket and falls back to `0.0.0.0` only on a host without IPv6 (see /// `add_wildcard_addrs_to_server` in `contrib/grpc`). A wildcard listener therefore overlaps every /// listener on that port, not only another wildcard listener. Normalize the gRPC-based listeners to /// the first wildcard address so a configuration such as `127.0.0.1`, `0.0.0.0`, `::1` creates one /// listener instead of creating overlapping sockets. This applies independently to each gRPC-based /// port because all such ports use the same `listen_host` configuration. Strings getGRPCListenHosts(const Strings & listen_hosts) { const auto wildcard_it = std::ranges::find_if(listen_hosts, isWildcardListenHost); if (wildcard_it == listen_hosts.end()) return listen_hosts; return {*wildcard_it}; } #endif } void Server::createServer( Poco::Util::AbstractConfiguration & config, const std::string & listen_host, const char * port_name, bool listen_try, bool start_server, std::vector & servers, CreateServerFunc && func) const { if (DB::createServer(config, listen_host, port_name, listen_try, start_server, servers, std::move(func), &logger()) && (start_server || !servers.back().bindsOnStart())) { /// Register the configured port rather than the actual bound port. `getServerPort` keeps a /// single value per `port_name`, so with `tcp_port=0` (OS-assigned) and several `listen_host` /// values (e.g. the default `::1` + `127.0.0.1`) each host binds a distinct ephemeral port and /// registering the actual port would let the last host overwrite the others, leaving /// `getServerPort` pointing at a port that is not listening on the host a client uses. When the /// configured port is non-zero it equals the bound port anyway, so this preserves the previous /// behavior in all cases. (`clickhouse-local` registers the actual bound port because it needs /// the OS-assigned value, but it rejects the ambiguous `port=0` + multiple `listen_host` combo.) global_context->registerServerPort(port_name, static_cast(config.getInt(port_name))); } } #if defined(OS_LINUX) namespace { void setOOMScore(int value, LoggerRawPtr log) { try { std::string value_string = std::to_string(value); DB::WriteBufferFromFile buf("/proc/self/oom_score_adj"); buf.write(value_string.c_str(), value_string.size()); buf.next(); buf.close(); } catch (const Poco::Exception & e) { LOG_WARNING(log, "Failed to adjust OOM score: '{}'.", e.displayText()); return; } LOG_INFO(log, "Set OOM score adjustment to {}", value); } } #endif void Server::uninitialize() { logger().information("shutting down"); BaseDaemon::uninitialize(); } int Server::run() { if (config().hasOption("help")) { Poco::Util::HelpFormatter help_formatter(Server::options()); std::string app_name = (commandName() == "clickhouse-server") ? "clickhouse-server" : "clickhouse server"; auto header_str = fmt::format("{} [OPTION] [-- [ARG]...]\n" "positional arguments can be used to rewrite config.xml properties, for example, --http_port=8010", app_name); help_formatter.setHeader(header_str); help_formatter.format(std::cout); return 0; } if (config().hasOption("version")) { std::cout << VERSION_NAME << " server version " << VERSION_STRING << VERSION_OFFICIAL << "." << std::endl; return 0; } return Application::run(); // NOLINT } void Server::initialize(Poco::Util::Application & self) { Poco::AutoPtr properties(new Poco::Util::MapConfiguration); Poco::Util::AbstractConfiguration::Keys property_keys; config().keys("application", property_keys); for (const auto & key : property_keys) properties->setString("application." + key, config().getRawString("application." + key)); application_properties = properties; ConfigProcessor::registerEmbeddedConfig("config.xml", std::string_view(reinterpret_cast(resource_embedded_xml), std::size(resource_embedded_xml))); BaseDaemon::initialize(self); logger().information("starting up"); LOG_INFO(&logger(), "OS name: {}, version: {}, architecture: {}", Poco::Environment::osName(), Poco::Environment::osVersion(), Poco::Environment::osArchitecture()); } std::string Server::getDefaultCorePath() const { return getCanonicalPath(config().getString("path", DBMS_DEFAULT_PATH), original_working_directory) + "cores"; } void Server::defineOptions(Poco::Util::OptionSet & options) { options.addOption( Poco::Util::Option("help", "h", "show help and exit") .required(false) .repeatable(false) .binding("help")); options.addOption( Poco::Util::Option("version", "V", "show version and exit") .required(false) .repeatable(false) .binding("version")); BaseDaemon::defineOptions(options); } [[maybe_unused]] static void checkForUsersNotInMainConfig( const Poco::Util::AbstractConfiguration & config, const ServerSettings & server_settings, const std::string & config_path, const std::string & users_config_path, LoggerPtr log) { if (server_settings[ServerSetting::skip_check_for_incorrect_settings]) return; if (config.has("users") || config.has("profiles") || config.has("quotas")) { /// We cannot throw exception here, because we have support for obsolete 'conf.d' directory /// (that does not correspond to config.d or users.d) but substitute configuration to both of them. LOG_ERROR(log, "The , and elements should be located in users config file: {} not in main config {}." " Also note that you should place configuration changes to the appropriate *.d directory like 'users.d'.", users_config_path, config_path); } } namespace { void loadStartupScripts(const Poco::Util::AbstractConfiguration & config, const ServerSettings & server_settings, ContextMutablePtr context, Poco::Logger * log) { try { Poco::Util::AbstractConfiguration::Keys keys; config.keys("startup_scripts", keys); SetResultDetailsFunc callback; std::vector skipped_startup_scripts; for (const auto & key : keys) { if (key == "throw_on_error") continue; std::string full_prefix = "startup_scripts." + key; auto user = config.getString(full_prefix + ".user", ""); auto startup_context = Context::createCopy(context); if (!user.empty()) { auto & access_control = startup_context->getAccessControl(); startup_context->setUser(access_control.getID(user)); } startup_context->makeQueryContext(); if (config.has(full_prefix + ".condition")) { auto condition = config.getString(full_prefix + ".condition"); auto condition_read_buffer = ReadBufferFromString(condition); auto condition_write_buffer = WriteBufferFromOwnString(); LOG_DEBUG(log, "Checking startup query condition `{}`", formatQueryForLogging(condition, startup_context->getSettingsRef())); startup_context->setQueryKind(ClientInfo::QueryKind::INITIAL_QUERY); startup_context->setCurrentQueryId(""); { auto query_scope = QueryScope::create(startup_context); executeQuery(condition_read_buffer, condition_write_buffer, startup_context, callback, QueryFlags{ .internal = true }, std::nullopt, {}); } auto result = condition_write_buffer.str(); if (result != "1\n" && result != "true\n") { if (result != "0\n" && result != "false\n") { if (result.empty()) LOG_DEBUG(log, "Skipping startup script as condition query returned empty value."); else LOG_DEBUG( log, "Skipping startup script as condition query returned value `{}` " "which can't be interpreted as a boolean (`0`, `false`, `1`, `true`).", result); skipped_startup_scripts.emplace_back(full_prefix); } continue; } LOG_DEBUG(log, "Condition is true, will execute the query next"); } auto query = config.getString(full_prefix + ".query"); auto read_buffer = ReadBufferFromString(query); auto write_buffer = WriteBufferFromOwnString(); LOG_DEBUG(log, "Executing query `{}`", formatQueryForLogging(query, startup_context->getSettingsRef())); startup_context->setQueryKind(ClientInfo::QueryKind::INITIAL_QUERY); startup_context->setCurrentQueryId(""); auto query_scope = QueryScope::create(startup_context); executeQuery(read_buffer, write_buffer, startup_context, callback, QueryFlags{ .internal = true }, std::nullopt, {}); } if (!skipped_startup_scripts.empty()) { context->addOrUpdateWarningMessage( Context::WarningType::SKIPPING_CONDITION_QUERY, PreformattedMessage::create( "Skipped the following startup script(s): {} as the condition query for those returned values, " "which can't be interpreted as a boolean (`0`, `false`, `1`, `true`).", fmt::join(skipped_startup_scripts, ", "))); } CurrentMetrics::set(CurrentMetrics::StartupScriptsExecutionState, StartupScriptsExecutionState::Success); } catch (...) { DimensionalMetrics::set( DimensionalMetrics::StartupScriptsFailureReason, {String(ErrorCodes::getName(getCurrentExceptionCode()))}, 1.0); CurrentMetrics::set(CurrentMetrics::StartupScriptsExecutionState, StartupScriptsExecutionState::Failure); tryLogCurrentException(log, "Failed to parse startup scripts file"); if (server_settings[ServerSetting::startup_scripts_throw_on_error]) throw Exception( ErrorCodes::STARTUP_SCRIPTS_ERROR, "Cannot finish startup_script successfully. Use startup_scripts.throw_on_error setting to change this behavior"); } } void initializeAzureSDKLogger( [[ maybe_unused ]] const ServerSettings & server_settings, [[ maybe_unused ]] int server_logs_level) { #if USE_AZURE_BLOB_STORAGE if (!server_settings[ServerSetting::enable_azure_sdk_logging]) return; using AzureLogsLevel = Azure::Core::Diagnostics::Logger::Level; static const std::unordered_map> azure_to_server_mapping = { {AzureLogsLevel::Error, {Poco::Message::PRIO_DEBUG, LogsLevel::debug}}, {AzureLogsLevel::Warning, {Poco::Message::PRIO_DEBUG, LogsLevel::debug}}, {AzureLogsLevel::Informational, {Poco::Message::PRIO_TRACE, LogsLevel::trace}}, {AzureLogsLevel::Verbose, {Poco::Message::PRIO_TEST, LogsLevel::test}}, }; static const std::map server_to_azure_mapping = { {Poco::Message::PRIO_DEBUG, AzureLogsLevel::Warning}, {Poco::Message::PRIO_TRACE, AzureLogsLevel::Informational}, {Poco::Message::PRIO_TEST, AzureLogsLevel::Verbose}, }; static const LoggerPtr azure_sdk_logger = getLogger("AzureSDK"); auto it = server_to_azure_mapping.lower_bound(static_cast(server_logs_level)); chassert(it != server_to_azure_mapping.end()); Azure::Core::Diagnostics::Logger::SetLevel(it->second); Azure::Core::Diagnostics::Logger::SetListener([](AzureLogsLevel level, const std::string & message) { auto [poco_level, db_level] = azure_to_server_mapping.at(level); LOG_IMPL(azure_sdk_logger, db_level, poco_level, fmt::runtime(message)); }); #endif } } namespace { /// Defined next to `resolveHTTPHandlersKey` below, which they walk the `impl` chain with; declared here /// because the configuration reload callback of `Server::main` validates a configuration with them. Strings servedHTTPHandlersKeys(const Poco::Util::AbstractConfiguration & config); bool hasPrometheusListener(const Poco::Util::AbstractConfiguration & config); } int Server::main(const std::vector & /*args*/) try { #if USE_SSH && defined(OS_LINUX) ::ssh::LibSSHInitializer::instance(); ::ssh::libsshLogger::initialize(); #endif Stopwatch startup_watch; Poco::Logger * log = &logger(); const size_t physical_server_memory = getMemoryAmount(); LOG_INFO( log, "Available RAM: {}; logical cores: {}; used cores: {}.", formatReadableSizeWithBinarySuffix(physical_server_memory), std::thread::hardware_concurrency(), getNumberOfCPUCoresToUse() // on ARM processors it can show only enabled at current moment cores ); #if defined(__x86_64__) String cpu_info; #define COLLECT_FLAG(X) \ if (CPU::have##X()) \ { \ if (!cpu_info.empty()) \ cpu_info += ", "; \ cpu_info += #X; \ } CPU_ID_ENUMERATE(COLLECT_FLAG) #undef COLLECT_FLAG LOG_INFO(log, "Available CPU instruction sets: {}", cpu_info); #endif ServerSettings server_settings; server_settings.loadSettingsFromConfig(config()); /// Fail closed if an operator is still on a legacy insert deduplication version. This build only /// writes the unified deduplication hash, so silently ignoring the setting would break the /// mixed-version deduplication contract; refuse to start and explain how to migrate. This must run /// after every server-settings load, not just this first one: the ZooKeeper-include reload below /// and the runtime config reloader re-read the settings, so a legacy value could otherwise slip in /// through a ZK-backed config or a `SYSTEM RELOAD CONFIG`. auto validate_insert_deduplication_version = [](const ServerSettings & settings) { const String dedup_version = settings[ServerSetting::insert_deduplication_version]; if (dedup_version != "new_unified_hash") throw Exception( ErrorCodes::BAD_ARGUMENTS, "Server setting 'insert_deduplication_version' is set to '{}', but this version of ClickHouse " "supports only the unified insert deduplication hash ('new_unified_hash'). Remove the setting from " "the configuration (or set it to 'new_unified_hash'). To migrate from a version that used " "'old_separate_hashes' or 'compatible_double_hashes', first run on a release that supports " "'compatible_double_hashes' (writing both the legacy and unified hashes), then upgrade to this " "version. For replicated tables run it for at least 'replicated_deduplication_window_seconds' " "(one hour by default); the default windows retain the unified hashes of all inserts for that " "one-hour window, which is considered long enough to cover an insert retry loop. For " "non-replicated tables with 'non_replicated_deduplication_window' > 0 that window is count-based, " "not time-based, so run 'compatible_double_hashes' for at least that many inserts before " "upgrading; otherwise stale legacy block ids can survive the upgrade and a retried insert may be " "accepted as new.", dedup_version); }; validate_insert_deduplication_version(server_settings); #if defined(OS_LINUX) std::string executable_path = getExecutablePath(); /// Remap before creating other threads to prevent crashes if (server_settings[ServerSetting::remap_executable]) { LOG_DEBUG(log, "Will remap executable in memory."); /// remapExecutable rewrites the whole code segment in place; any other thread executing code during /// the window faults (and the signal handler's code is unmapped too, so it dies silently). /// /// Restart the async logging threads even if remapExecutable throws, so the exception unwinding /// through Server::main is logged by a fully running logger (a stopped channel falls back to /// synchronous delivery, but anything already queued waits until the threads run again). /// The remap exception is stashed rather than rethrown directly, so the restart happens outside the /// signal-blocking scope: the mask is restored first, and the logging threads get the same mask as /// during a normal startup. std::exception_ptr remap_exception; size_t remapped_size = 0; #if USE_JEMALLOC bool jemalloc_background_threads_were_enabled = false; #endif { /// `BaseDaemon::initializeTerminationAndSignalProcessing` has already installed the signal handlers /// and started the signal listener thread. Block the asynchronously delivered handled signals in this /// thread first, so that no new record can be queued into the signal pipe from here on. Nothing is /// lost: the signal stays pending for the process and is delivered once the mask is restored below. BlockSignalsScope block_signals(asynchronousHandledSignals()); #if USE_JEMALLOC /// `Jemalloc::setup` in `BaseDaemon::initialize` has already started jemalloc background threads /// (`jemalloc_enable_background_threads` defaults to true). They execute allocator code from the text /// segment, and they were created with an unblocked signal mask, so they could also take a handled /// signal inside the remap window. Writing `false` into the `background_thread` mallctl stops *and /// joins* every background thread before returning, so after this call they are fully quiesced. /// They are restarted below, after the remap. `max_background_threads` is a separate mallctl and is /// left untouched, so re-enabling restores the configured limit. if (Jemalloc::tryGetValue("background_thread", jemalloc_background_threads_were_enabled) && jemalloc_background_threads_were_enabled) Jemalloc::setBackgroundThreads(false); #endif /// Blocking the signals is not enough by itself: a signal that arrived just before the mask was set /// may already have written a record into the signal pipe, and the listener thread would execute the /// corresponding callback at an arbitrary later moment - possibly inside the remap window. Stop and /// join the listener thread: it drains every record already queued (they are ordered in the pipe /// before the stop request) while the code is still mapped, and then exits. It is restarted below. stopSignalListener(); /// The async logging threads poll rather than block, so join them for the duration and restart afterwards. /// This fails closed: if any logging thread cannot be joined, the exception propagates and the remap /// is aborted, instead of rewriting the text segment under a thread that may still execute from it. stopAsyncLoggingThreads(); /// At this point the process is single-threaded again and all handled signals are blocked, /// so no code other than this thread can possibly run while the text segment is rewritten. try { remapped_size = remapExecutable(); } catch (...) { remap_exception = std::current_exception(); } } #if USE_JEMALLOC /// Restarted outside the signal-blocking scope, so the new background threads inherit the same /// (unblocked) signal mask they had before the remap. Restarted even if remapExecutable threw, /// for the same reason the logging threads are: the allocator must be back in its configured /// state while the exception unwinds. `verifySetup` below cross-checks the state against the /// server settings, so a failure to restore would not go unnoticed. if (jemalloc_background_threads_were_enabled) Jemalloc::setBackgroundThreads(true); #endif /// Fail closed if the restart itself throws: continuing with only part of the logging threads /// running would keep accepting messages into queues that no consumer drains. open() tears the /// partially opened channel back down before rethrowing, and a stopped channel delivers messages /// synchronously, so the exception propagating out of Server::main still reaches the log. startAsyncLoggingThreads(); /// Restarted after the logging threads, so that if this throws, the exception is still logged normally. /// A pending signal delivered when the mask was restored above only writes a record into the signal /// pipe; the record waits there until the restarted listener thread picks it up, so it is not lost. startSignalListener(); if (remap_exception) std::rethrow_exception(remap_exception); LOG_DEBUG(log, "The code ({}) in memory has been successfully remapped.", ReadableSize(remapped_size)); } if (server_settings[ServerSetting::mlock_executable]) { size_t min_physical_server_memory_to_mlock = server_settings[ServerSetting::mlock_executable_min_total_memory_amount_bytes]; if (physical_server_memory >= min_physical_server_memory_to_mlock) { if (hasLinuxCapability(CAP_IPC_LOCK)) { try { /// Get the memory area with (current) code segment. /// It's better to lock only the code segment instead of calling "mlockall", /// because otherwise debug info will be also locked in memory, and it can be huge. auto [addr, len] = getMappedArea(reinterpret_cast(mainEntryClickHouseServer)); LOG_TRACE(log, "Will do mlock to prevent executable memory from being paged out. It may take a few seconds."); if (0 != mlock(addr, len)) LOG_WARNING(log, "Failed mlock: {}", errnoToString()); else LOG_TRACE(log, "The memory map of clickhouse executable has been mlock'ed, total {}", ReadableSize(len)); } catch (...) { LOG_WARNING(log, "Cannot mlock: {}", getCurrentExceptionMessage(false)); } } else { LOG_INFO( log, "It looks like the process has no CAP_IPC_LOCK capability, binary mlock will be disabled." " It could happen due to incorrect ClickHouse package installation." " You could resolve the problem manually with 'sudo setcap cap_ipc_lock=+ep {}'." " Note that it will not work on 'nosuid' mounted filesystems.", executable_path.empty() ? "/usr/bin/clickhouse" : executable_path); } } else { LOG_INFO(log, "Skip mlock for the clickhouse executable, because the total memory in the system ({}) is less than the minimum configured threshold (`mlock_executable_min_total_memory_amount_bytes` = {})", ReadableSize(physical_server_memory), ReadableSize(min_physical_server_memory_to_mlock)); } } #endif // If the startup_level is set in the config, we override the root logger level. // Specific loggers can still override it. std::string default_logger_level_config = server_settings[ServerSetting::logger_level]; bool should_restore_default_logger_level = false; const auto & logger_startup_level_setting = server_settings[ServerSetting::logger_startup_level]; if (logger_startup_level_setting.changed && !logger_startup_level_setting.value.empty()) { /// Set the root logger level to the startup level. /// This is useful for debugging startup issues. /// The root logger level will be reset to the default level after the server is fully initialized. config().setString("logger.level", logger_startup_level_setting.value); Loggers::updateLevels(config(), logger()); should_restore_default_logger_level = true; LOG_INFO(log, "Starting root logger in level {}", logger_startup_level_setting.value); } // If the startup_console_log_level is set in the config, we override the console logger level. // Specific loggers can still override it. bool console_log_level_was_set = config().has("logger.console_log_level"); std::string original_console_log_level_config = config().getString("logger.console_log_level", ""); bool should_restore_console_log_level = false; if (config().has("logger.startup_console_log_level") && !config().getString("logger.startup_console_log_level").empty()) { /// Set the console logger level to the startup console level. /// This is useful for debugging startup issues. /// The console logger level will be reset to the default level after the server is fully initialized. config().setString("logger.console_log_level", config().getString("logger.startup_console_log_level")); Loggers::updateLevels(config(), logger()); should_restore_console_log_level = true; LOG_INFO(log, "Starting console logger in level {}", config().getString("logger.startup_console_log_level")); } MainThreadStatus::getInstance(); #if USE_JEMALLOC Jemalloc::verifySetup( server_settings[ServerSetting::jemalloc_enable_global_profiler], server_settings[ServerSetting::jemalloc_enable_background_threads], server_settings[ServerSetting::jemalloc_max_background_threads_num], server_settings[ServerSetting::jemalloc_collect_global_profile_samples_in_trace_log], server_settings[ServerSetting::jemalloc_profiler_sampling_rate]); #endif StackTrace::setShowAddresses(server_settings[ServerSetting::show_addresses_in_stack_traces]); ProfileEvents::setUserPerCPUEnabled(server_settings[ServerSetting::user_profile_events_per_cpu]); #if USE_HDFS /// This will point libhdfs3 to the right location for its config. /// Note: this has to be done once at server initialization, because 'setenv' is not thread-safe. String libhdfs3_conf = server_settings[ServerSetting::hdfs_libhdfs3_conf]; if (!libhdfs3_conf.empty()) { if (std::filesystem::path{libhdfs3_conf}.is_relative() && !std::filesystem::exists(libhdfs3_conf)) { const String config_path = server_settings[ServerSetting::config_file]; const auto config_dir = std::filesystem::path{config_path}.remove_filename(); if (std::filesystem::exists(config_dir / libhdfs3_conf)) libhdfs3_conf = std::filesystem::absolute(config_dir / libhdfs3_conf); } setenv("LIBHDFS3_CONF", libhdfs3_conf.c_str(), true /* overwrite */); // NOLINT } #endif if (auto total_numa_memory = getNumaNodesTotalMemory(); total_numa_memory.has_value()) { LOG_INFO( log, "ClickHouse is bound to a subset of NUMA nodes. Total memory of all available nodes: {}", ReadableSize(*total_numa_memory)); } #if USE_AWS_S3 DB::S3::setCredentialsProviderCacheMaxSize(server_settings[ServerSetting::s3_credentials_provider_max_cache_size]); #endif registerInterpreters(); registerParsers(); registerFunctions(); registerAggregateFunctions(); registerTableFunctions(); registerDatabases(); registerStorages(); registerDictionaries(); setSecretArgumentsFinder(&SecretArgumentsRegistry::instance()); registerDisks(/* global_skip_access_check= */ false); registerFormats(); registerRemoteFileMetadatas(); QueryPlanStepRegistry::registerPlanSteps(); CurrentMetrics::set(CurrentMetrics::Revision, ClickHouseRevision::getVersionRevision()); CurrentMetrics::set(CurrentMetrics::VersionInteger, ClickHouseRevision::getVersionInteger()); /** Context contains all that query execution is dependent: * settings, available functions, data types, aggregate functions, databases, ... */ auto shared_context = Context::createShared(); global_context = Context::createGlobal(shared_context.get()); global_context->makeGlobalContext(); global_context->setApplicationType(Context::ApplicationType::SERVER); ThreadFuzzer::instance().setup(); addBuildWarnings(global_context); /// Under thread sanitizer we use frame-pointer-based unwinding (via abseil) which does not /// call dl_iterate_phdr in the signal handler, so the PHDR cache is not needed. #if defined(THREAD_SANITIZER) bool has_trace_collector = config().has("trace_log"); LOG_INFO(log, "Query Profiler will use frame-pointer-based stack unwinding under sanitizers."); #else bool has_trace_collector = hasAsyncSignalSafeUnwind() && config().has("trace_log"); if (!hasAsyncSignalSafeUnwind()) LOG_INFO(log, "Query Profiler and TraceCollector are disabled because async-signal-safe stack unwinding" " is not available in this build (on Linux this requires the lock-free PHDR cache, otherwise" " 'dl_iterate_phdr' is not lock free and not async-signal safe)."); #endif #if defined(MEMORY_SANITIZER) /// The TraceCollector itself stays enabled: the memory profiler, `trace_profile_events` and /// `SYSTEM INSTRUMENT` all feed it from ordinary code rather than from a signal handler. LOG_INFO(log, "The sampling Query Profiler is disabled under Memory Sanitizer, because a profiler signal" " delivered to a thread that is printing a sanitizer report aborts the process and truncates the" " report. See QUERY_PROFILER_SUPPORTED in Common/QueryProfiler.h."); #endif // Settings validation for page cache. Ensure that page_cache_max_size is > page_cache_min_size. // Otherwise, crash might happen during cache resizing in src/Common/PageCache.cpp::autoResize size_t page_cache_min_size = server_settings[ServerSetting::page_cache_min_size]; size_t page_cache_max_size = server_settings[ServerSetting::page_cache_max_size]; if (page_cache_max_size != 0 && (page_cache_min_size > page_cache_max_size)) { throw Exception( ErrorCodes::INVALID_CONFIG_PARAMETER, "Invalid page cache configuration: page_cache_min_size ({}) is greater than page_cache_max_size ({}).", page_cache_min_size, page_cache_max_size); } size_t update_period_seconds = server_settings[ServerSetting::asynchronous_metrics_update_period_s]; size_t heavy_metrics_update_period_seconds = server_settings[ServerSetting::asynchronous_heavy_metrics_update_period_s]; if (update_period_seconds == 0 || heavy_metrics_update_period_seconds == 0) { throw Exception(ErrorCodes::INVALID_SETTING_VALUE, "Settings asynchronous_metrics_update_period_s and asynchronous_heavy_metrics_update_period_s must not be zero"); } // Initialize global thread pool. Do it before we fetch configs from zookeeper // nodes (`from_zk`), because ZooKeeper interface uses the pool. We will // ignore `max_thread_pool_size` in configs we fetch from ZK, but oh well. GlobalThreadPool::initialize( server_settings[ServerSetting::max_thread_pool_size], server_settings[ServerSetting::max_thread_pool_free_size], server_settings[ServerSetting::thread_pool_queue_size], has_trace_collector ? server_settings[ServerSetting::global_profiler_real_time_period_ns].value : 0, has_trace_collector ? server_settings[ServerSetting::global_profiler_cpu_time_period_ns].value : 0); #if USE_SILK if (server_settings[ServerSetting::enable_silk_runtime]) { Silk::initializeFiberScheduler(config().getUInt("silk.fiber_stack_size", Silk::DEFAULT_FIBER_STACK_SIZE)); } #endif if (has_trace_collector) { global_context->createTraceCollector(); /// The config reloader applies this too; the seed here covers startup, which runs before its first callback. MemoryTracker::setMinAllocationSizeToLogStackTrace( server_settings[ServerSetting::min_allocation_size_to_log_stack_trace]); /// Set up server-wide memory profiler (for total memory tracker). if (server_settings[ServerSetting::total_memory_profiler_step]) total_memory_tracker.setProfilerStep(server_settings[ServerSetting::total_memory_profiler_step]); if (server_settings[ServerSetting::total_memory_tracker_sample_probability] > 0.0) total_memory_tracker.setSampleProbability(server_settings[ServerSetting::total_memory_tracker_sample_probability]); if (server_settings[ServerSetting::total_memory_profiler_sample_min_allocation_size]) total_memory_tracker.setSampleMinAllocationSize(server_settings[ServerSetting::total_memory_profiler_sample_min_allocation_size]); if (server_settings[ServerSetting::total_memory_profiler_sample_max_allocation_size]) total_memory_tracker.setSampleMaxAllocationSize(server_settings[ServerSetting::total_memory_profiler_sample_max_allocation_size]); if (current_thread) current_thread->resolveMemorySampleConfig(); } total_memory_tracker.setJemallocFlushProfileInterval(server_settings[ServerSetting::jemalloc_flush_profile_interval_bytes]); total_memory_tracker.setJemallocFlushProfileOnMemoryExceeded(server_settings[ServerSetting::jemalloc_flush_profile_on_memory_exceeded]); total_memory_tracker.setJemallocFlushProfileOnMemoryExceededSeconds(server_settings[ServerSetting::jemalloc_flush_profile_on_memory_exceeded_interval]); Poco::ThreadPool server_pool( /* minCapacity */3, /* maxCapacity */server_settings[ServerSetting::max_connections], /* idleTime */60, /* stackSize */DEFAULT_THREAD_STACK_SIZE ? static_cast(DEFAULT_THREAD_STACK_SIZE) : POCO_THREAD_STACK_SIZE, server_settings[ServerSetting::global_profiler_real_time_period_ns], server_settings[ServerSetting::global_profiler_cpu_time_period_ns]); std::unique_ptr introspection_server_pool; std::mutex servers_lock; std::vector servers; std::vector servers_to_start_before_tables; std::vector introspection_servers; auto stop_silk_fiber_scheduler = [&]{ #if USE_SILK if (server_settings[ServerSetting::enable_silk_runtime]) { LOG_INFO(log, "Stopping silk fiber scheduler"); Silk::destroyFiberScheduler(); } #endif }; /// Wait for all threads to avoid possible use-after-free (for example logging objects can be already destroyed). SCOPE_EXIT_SAFE({ Stopwatch watch; LOG_INFO(log, "Waiting for background threads"); DB::StaticThreadPool::shutdownAll(); GlobalThreadPool::instance().shutdown(); LOG_INFO(log, "Background threads finished in {} ms", watch.elapsedMilliseconds()); stop_silk_fiber_scheduler(); }); if (page_cache_max_size != 0) { global_context->setPageCache( std::chrono::milliseconds(Int64(server_settings[ServerSetting::page_cache_history_window_ms])), server_settings[ServerSetting::page_cache_policy], server_settings[ServerSetting::page_cache_size_ratio], server_settings[ServerSetting::page_cache_min_size], server_settings[ServerSetting::page_cache_max_size], server_settings[ServerSetting::page_cache_free_memory_ratio], server_settings[ServerSetting::page_cache_shards]); total_memory_tracker.setPageCache(global_context->getPageCache().get()); } MemoryWorkerConfig memory_worker_config{ .rss_update_period_ms = server_settings[ServerSetting::memory_worker_period_ms], .purge_dirty_pages_threshold_ratio = server_settings[ServerSetting::memory_worker_purge_dirty_pages_threshold_ratio], .purge_total_memory_threshold_ratio = server_settings[ServerSetting::memory_worker_purge_total_memory_threshold_ratio], .correct_tracker = server_settings[ServerSetting::memory_worker_correct_memory_tracker], .decay_adjustment_period_ms = server_settings[ServerSetting::memory_worker_decay_adjustment_period_ms], .use_cgroup = server_settings[ServerSetting::memory_worker_use_cgroup], .rss_speculative_reserve_ratio = server_settings[ServerSetting::memory_worker_rss_speculative_reserve_ratio], .dynamic_hard_limit_ratio = server_settings[ServerSetting::memory_worker_dynamic_hard_limit] ? static_cast(server_settings[ServerSetting::max_server_memory_usage_to_ram_ratio]) : 0.0, }; MemoryWorker memory_worker(memory_worker_config, global_context->getPageCache()); /// This object will periodically calculate some metrics. std::unique_ptr async_metrics; DB::AsynchronousMetrics::ProtocolServerMetricsFunc asynchronous_metrics_protocol_server_metrics_func = [&]() -> std::vector { std::vector metrics; std::lock_guard lock(servers_lock); metrics.reserve(servers_to_start_before_tables.size() + servers.size() + introspection_servers.size()); for (const auto & server : servers_to_start_before_tables) metrics.emplace_back(ProtocolServerMetrics{ server.getPortName(), server.getProtocolType(), server.currentThreads(), server.refusedConnections()}); for (const auto & server : servers) metrics.emplace_back(ProtocolServerMetrics{ server.getPortName(), server.getProtocolType(), server.currentThreads(), server.refusedConnections()}); for (const auto & server : introspection_servers) metrics.emplace_back(ProtocolServerMetrics{ server.getPortName(), server.getProtocolType(), server.currentThreads(), server.refusedConnections()}); return metrics; }; const unsigned async_metrics_update_period_s = server_settings[ServerSetting::asynchronous_metrics_update_period_s]; const bool async_metrics_update_jemalloc_epoch = memory_worker.getSource() != MemoryWorker::MemoryUsageSource::Jemalloc; const bool async_metrics_update_rss = memory_worker.getSource() == MemoryWorker::MemoryUsageSource::None; #if USE_NURAFT const bool async_metrics_keeper_metrics_only = server_settings[ServerSetting::asynchronous_metrics_keeper_metrics_only]; if (async_metrics_keeper_metrics_only) { async_metrics = std::make_unique( global_context, async_metrics_update_period_s, asynchronous_metrics_protocol_server_metrics_func, async_metrics_update_jemalloc_epoch, async_metrics_update_rss ); } else #endif { async_metrics = std::make_unique( global_context, async_metrics_update_period_s, server_settings[ServerSetting::asynchronous_metrics_enable_heavy_metrics], server_settings[ServerSetting::asynchronous_heavy_metrics_update_period_s], asynchronous_metrics_protocol_server_metrics_func, async_metrics_update_jemalloc_epoch, async_metrics_update_rss ); } auto begin_keeper_tcp_connection_drain = [&] { #if USE_NURAFT if (auto dispatcher = global_context->tryGetKeeperDispatcher()) dispatcher->beginTCPConnectionDrain(); KeeperTCPHandler::closeAllConnections(); #endif }; auto is_keeper_tcp_server = [](const ProtocolServerAdapter & server) { const auto & port_name = server.getPortName(); return port_name == "keeper_server.tcp_port" || port_name == "keeper_server.tcp_port_secure"; }; /// NOTE: global context should be destroyed *before* GlobalThreadPool::shutdown() /// Otherwise GlobalThreadPool::shutdown() will hang, since Context holds some threads. SCOPE_EXIT_SAFE({ /// Required for the startup exception case where the termination block that should set is_cancelled=true never runs. is_cancelled = true; /// Stop accepting connections on the regular servers. In the normal shutdown path they are /// already stopped, but on startup failure some of them can still be running: the Prometheus /// endpoint is started before tables are loaded. Otherwise `server_pool.joinAll()` below /// would wait forever for their listener threads. { std::lock_guard lock(servers_lock); for (auto & server : servers) if (!server.isStopping()) server.stop(); } async_metrics->stop(); /** Ask to cancel background jobs all table engines, * and also query_log. * It is important to do early, not in destructor of Context, because * table engines could use Context on destroy. */ LOG_INFO(log, "Shutting down storages."); /// The columns cache goes away with the context; stop resizing it. setMemoryReleasableCache(nullptr); memory_worker.setReleasableCache(nullptr); global_context->shutdown(); LOG_DEBUG(log, "Shut down storages."); size_t keeper_tcp_connections = 0; size_t non_keeper_tcp_connections = 0; if (!servers_to_start_before_tables.empty()) { LOG_DEBUG(log, "Waiting for current connections to servers for tables to finish."); { std::lock_guard lock(servers_lock); for (auto & server : servers_to_start_before_tables) { server.stop(); if (is_keeper_tcp_server(server)) keeper_tcp_connections += server.currentConnections(); else non_keeper_tcp_connections += server.currentConnections(); } } } /// Stop Keeper TCP handlers before draining the remaining pre-table protocol handlers. /// Those handlers include HTTP control, interserver HTTP/HTTPS, and Prometheus; they need /// the embedded Keeper and RAFT to remain live until they finish. begin_keeper_tcp_connection_drain(); if (non_keeper_tcp_connections) { LOG_INFO(log, "Closed all non-Keeper-TCP listening sockets. Waiting for {} outstanding connections.", non_keeper_tcp_connections); non_keeper_tcp_connections = waitServersToFinish( servers_to_start_before_tables, servers_lock, server_settings[ServerSetting::shutdown_wait_unfinished], [&](const auto & server) { return !is_keeper_tcp_server(server); }); } global_context->signalKeeperDispatcherShutdown(); global_context->shutdownKeeperDispatcherBeforeConnectionsFinish(); if (non_keeper_tcp_connections) { global_context->shutdownKeeperDispatcherAfterConnectionsFinish(false); dumpCoverageReportIfPossible(); LOG_WARNING( log, "Closed connections to non-Keeper-TCP servers. But {} remain. Will shutdown forcefully.", non_keeper_tcp_connections); safeExit(0, LeakCheck::SkipAndReport); } if (!servers_to_start_before_tables.empty()) { if (keeper_tcp_connections) LOG_INFO(log, "Closed all Keeper TCP listening sockets. Waiting for {} outstanding connections.", keeper_tcp_connections); else LOG_INFO(log, "Closed all Keeper listening sockets."); if (keeper_tcp_connections > 0) keeper_tcp_connections = waitServersToFinish( servers_to_start_before_tables, servers_lock, server_settings[ServerSetting::shutdown_wait_unfinished], is_keeper_tcp_server); if (keeper_tcp_connections) LOG_INFO(log, "Closed Keeper TCP connections. But {} remain.", keeper_tcp_connections); else LOG_INFO(log, "Closed Keeper TCP connections."); } global_context->shutdownKeeperDispatcherAfterConnectionsFinish(keeper_tcp_connections == 0); /// Wait server pool to avoid use-after-free of destroyed context in the handlers server_pool.joinAll(); /** Explicitly destroy Context. It is more convenient than in destructor of Server, because logger is still available. * At this moment, no one could own shared part of Context. */ global_context->resetSharedContext(); global_context.reset(); shared_context.reset(); LOG_DEBUG(log, "Destroyed global context."); }); #if USE_AZURE_BLOB_STORAGE /// It makes sense to deinitialize libxml after joining of all threads /// in global pool because libxml uses thread-local memory allocations via /// 'pthread_key_create' and 'pthread_setspecific' which should be deallocated /// at 'pthread_exit'. Deinitialization of libxml leads to call of 'pthread_key_delete' /// and if it is done before joining of threads, allocated memory will not be freed /// and there may be memory leaks in threads that used libxml. GlobalThreadPool::instance().addOnDestroyCallback([] { Azure::Storage::_internal::XmlGlobalDeinitialize(); }); #endif getIOThreadPool().initialize( server_settings[ServerSetting::max_io_thread_pool_size], server_settings[ServerSetting::max_io_thread_pool_free_size], server_settings[ServerSetting::io_thread_pool_queue_size]); getBackupsIOThreadPool().initialize( server_settings[ServerSetting::max_backups_io_thread_pool_size], server_settings[ServerSetting::max_backups_io_thread_pool_free_size], server_settings[ServerSetting::backups_io_thread_pool_queue_size]); getFetchPartitionThreadPool().initialize( server_settings[ServerSetting::max_fetch_partition_thread_pool_size], 0, // FETCH PARTITION is relatively rare, no need to keep threads server_settings[ServerSetting::max_fetch_partition_thread_pool_size]); getActivePartsLoadingThreadPool().initialize( server_settings[ServerSetting::max_active_parts_loading_thread_pool_size], 0, // We don't need any threads once all the parts will be loaded server_settings[ServerSetting::max_active_parts_loading_thread_pool_size]); getOutdatedPartsLoadingThreadPool().initialize( server_settings[ServerSetting::max_outdated_parts_loading_thread_pool_size], 0, // We don't need any threads once all the parts will be loaded server_settings[ServerSetting::max_outdated_parts_loading_thread_pool_size]); /// It could grow if we need to synchronously wait until all the data parts will be loaded. getOutdatedPartsLoadingThreadPool().setMaxTurboThreads( server_settings[ServerSetting::max_active_parts_loading_thread_pool_size] ); getUnexpectedPartsLoadingThreadPool().initialize( server_settings[ServerSetting::max_unexpected_parts_loading_thread_pool_size], 0, // We don't need any threads once all the parts will be loaded server_settings[ServerSetting::max_unexpected_parts_loading_thread_pool_size]); /// It could grow if we need to synchronously wait until all the data parts will be loaded. getUnexpectedPartsLoadingThreadPool().setMaxTurboThreads( server_settings[ServerSetting::max_active_parts_loading_thread_pool_size] ); getPartsCleaningThreadPool().initialize( server_settings[ServerSetting::max_parts_cleaning_thread_pool_size], 0, // We don't need any threads one all the parts will be deleted server_settings[ServerSetting::max_parts_cleaning_thread_pool_size]); auto max_database_replicated_create_table_thread_pool_size = server_settings[ServerSetting::max_database_replicated_create_table_thread_pool_size] ? server_settings[ServerSetting::max_database_replicated_create_table_thread_pool_size] : getNumberOfCPUCoresToUse(); getDatabaseReplicatedCreateTablesThreadPool().initialize( max_database_replicated_create_table_thread_pool_size, 0, // We don't need any threads once all the tables will be created max_database_replicated_create_table_thread_pool_size); getDatabaseCatalogDropTablesThreadPool().initialize( server_settings[ServerSetting::database_catalog_drop_table_concurrency], 0, // We don't need any threads if there are no DROP queries. server_settings[ServerSetting::database_catalog_drop_table_concurrency]); /// Zero means the number of CPU cores. const size_t shutdown_concurrency = server_settings[ServerSetting::database_catalog_shutdown_table_concurrency] ? server_settings[ServerSetting::database_catalog_shutdown_table_concurrency] : getNumberOfCPUCoresToUse(); getDatabaseCatalogShutdownTablesThreadPool().initialize( shutdown_concurrency, 0, // Threads are only needed during server shutdown. shutdown_concurrency); getMergeTreePrefixesDeserializationThreadPool().initialize( server_settings[ServerSetting::max_prefixes_deserialization_thread_pool_size], server_settings[ServerSetting::max_prefixes_deserialization_thread_pool_free_size], server_settings[ServerSetting::prefixes_deserialization_thread_pool_thread_pool_queue_size]); getFormatParsingThreadPool().initialize( server_settings[ServerSetting::max_format_parsing_thread_pool_size], server_settings[ServerSetting::max_format_parsing_thread_pool_free_size], server_settings[ServerSetting::format_parsing_thread_pool_queue_size]); getAIRequestThreadPool().initialize( server_settings[ServerSetting::ai_function_max_thread_pool_size], server_settings[ServerSetting::ai_function_max_thread_pool_free_size], server_settings[ServerSetting::ai_function_thread_pool_queue_size]); std::string path_str = getCanonicalPath(String(server_settings[ServerSetting::path]), original_working_directory); fs::path path = path_str; /// Check that the process user id matches the owner of the data. assertProcessUserMatchesDataOwner( path_str, [&](const PreformattedMessage & message) { global_context->addOrUpdateWarningMessage( Context::WarningType::PROCESS_USER_MATCHES_DATA_OWNER, message); }); global_context->setPath(path_str); StatusFile status{path / "status", StatusFile::write_full_info}; ServerUUID::load(path / "uuid", log); PlacementInfo::PlacementInfo::instance().initialize(config()); zkutil::validateZooKeeperConfig(config()); bool has_zookeeper = zkutil::hasZooKeeperConfig(config()); if (has_zookeeper && config().getBool("keeper_server.standalone_keeper", false)) throw Exception(ErrorCodes::BAD_ARGUMENTS, "Cannot use standalone_keeper=true with a configured zookeeper client connection. " "A standalone keeper should not have a section in its config."); auto main_config_zk_node_cache = std::make_unique([&] { return global_context->getZooKeeper(); }); Coordination::EventPtr main_config_zk_changed_event = std::make_shared(); if (loaded_config.has_zk_includes) { ConfigProcessor config_processor(config_path); loaded_config = config_processor.loadConfigWithZooKeeperIncludes( main_config_zk_node_cache.get(), main_config_zk_changed_event, /* fallback_to_preprocessed = */ true); config_processor.savePreprocessedConfig(loaded_config, path_str); config().replace("default", loaded_config.configuration.duplicate(), PRIO_DEFAULT, false); } Settings::checkNoSettingNamesAtTopLevel(config(), config_path); /// Validate the loaded XML config directly (not the layered config) so that command-line /// options injected by `argsToConfig` (`--config-file`, `--daemon`, `-C`, ...) and Poco-internal /// layers (`system`, `application`) are not mistaken for unknown top-level config keys. /// The `skip_check_for_incorrect_settings` escape hatch is resolved from the layered `config()`, /// so it works when supplied from the command line as well as from the config file. ServerSettings::checkUnknownSettings( *loaded_config.configuration, config_path, config().getBool("skip_check_for_incorrect_settings", false)); /// We need to reload server settings because config could be updated via zookeeper. server_settings.loadSettingsFromConfig(config()); validate_insert_deduplication_version(server_settings); global_context->configureServerWideThrottling(); /// Create the dedicated MergeTree metadata arena pool. Placed after the ZooKeeper-include reload /// above so a `from_zk` value of the setting is honored, and still well before any parts are loaded. JemallocMergeTreeArena::initialize(server_settings[ServerSetting::jemalloc_merge_tree_arenas]); addMergeTreeArenaPoolWarnings(global_context); /// Restrict the server to the system calls it is known to use, as early in the startup as the /// configuration allows. That is after the ZooKeeper-include reload above, not before it: a /// filter cannot be removed or relaxed afterwards, so one installed from the configuration as it /// was before a `from_zk` value arrived would stay in force while `system.server_settings` /// reported the value from ZooKeeper. `TSYNC` extends the filter to the threads that already /// exist, so everything that runs from here on - including the processes the server forks /// later, which inherit it - is covered. const SeccompMode seccomp_mode = server_settings[ServerSetting::seccomp]; if (const SeccompFilterStatus seccomp_status = installSeccompFilter(seccomp_mode); seccomp_status.allowed_syscalls != 0) LOG_INFO( log, "Applied a seccomp policy to this process, allowing {} system calls. A system call outside the policy will " "be handled according to the `seccomp` server setting, which is set to `{}`", seccomp_status.allowed_syscalls, SettingFieldSeccompMode(seccomp_mode).toString()); else if (seccomp_mode != SeccompMode::Disabled) #if defined(OS_LINUX) LOG_WARNING( log, "The `seccomp` server setting is set to `{}`, but {}, so the server is running without a seccomp policy. " "`PR_SET_NO_NEW_PRIVS` has been set anyway, so nothing this process runs can gain privileges through a setuid program", SettingFieldSeccompMode(seccomp_mode).toString(), seccomp_status.not_installed_reason); #else LOG_WARNING( log, "The `seccomp` server setting is set to `{}`, but {}, so the server is running without a seccomp policy", SettingFieldSeccompMode(seccomp_mode).toString(), seccomp_status.not_installed_reason); #endif #if defined(OS_LINUX) if (server_settings[ServerSetting::skip_binary_checksum_checks]) { LOG_WARNING(log, "Binary checksum checks disabled due to skip_binary_checksum_checks - not recommended for production deployments"); } else if (!executable_path.empty()) { /// Integrity check based on checksum of the executable code. /// Note: it is not intended to protect from malicious party, /// because the reference checksum can be easily modified as well. /// And we don't involve asymmetric encryption with PKI yet. /// It's only intended to protect from faulty hardware. /// Note: it is only based on machine code. /// But there are other sections of the binary (e.g. exception handling tables) /// that are interpreted (not executed) but can alter the behaviour of the program as well. /// Please keep the below log messages in-sync with the ones in daemon/BaseDaemon.cpp if (stored_binary_hash.empty()) { LOG_WARNING(log, "Integrity check of the executable skipped because the reference checksum could not be read."); } else { String calculated_binary_hash = getHashOfLoadedBinaryHex(); if (calculated_binary_hash == stored_binary_hash) { LOG_INFO(log, "Integrity check of the executable successfully passed (checksum: {})", calculated_binary_hash); } else { /// Note: this must not rely on a system call the `seccomp` policy denies, such as /// `ptrace`, because the filter is already installed by this point. if (isRunUnderDebugger()) { /// Program is run under debugger. Modification of it's binary image is ok for breakpoints. global_context->addOrUpdateWarningMessage( Context::WarningType::SERVER_RUN_UNDER_DEBUGGER, PreformattedMessage::create( "Server is run under debugger and its binary image {} is modified (most likely with breakpoints).", calculated_binary_hash)); } else { throw Exception( ErrorCodes::CORRUPTED_DATA, "Calculated checksum of the executable ({0}) does not correspond" " to the reference checksum stored in the executable ({1})." " This may indicate one of the following:" " - the executable {2} was changed just after startup;" " - the executable {2} was corrupted on disk due to faulty hardware;" " - the loaded executable was corrupted in memory due to faulty hardware;" " - the file {2} was intentionally modified;" " - a logical error in the code.", calculated_binary_hash, stored_binary_hash, executable_path); } } } } /// Inject and enable failpoints declared in the configuration { auto & conf = config(); String root_key = "fail_points_active"; Poco::Util::AbstractConfiguration::Keys fail_point_names; conf.keys(root_key, fail_point_names); for (const auto & fail_point_name : fail_point_names) { if (ConfigHelper::getBool(conf, root_key + "." + fail_point_name)) FailPointInjection::enableFailPoint(fail_point_name); } } #endif memory_worker.start(); #if defined(OS_LINUX) int oom_score = server_settings[ServerSetting::oom_score]; if (oom_score) setOOMScore(oom_score, log); #endif #if defined(OS_LINUX) std::optional oom_canary; if (server_settings[ServerSetting::oom_canary_enable]) { OOMCanary::Config canary_config; canary_config.size_bytes = server_settings[ServerSetting::oom_canary_size]; canary_config.relaunch = server_settings[ServerSetting::oom_canary_relaunch]; canary_config.max_rapid_relaunches = server_settings[ServerSetting::oom_canary_max_rapid_relaunches]; canary_config.initial_backoff_seconds = server_settings[ServerSetting::oom_canary_initial_backoff_seconds]; canary_config.max_backoff_seconds = server_settings[ServerSetting::oom_canary_max_backoff_seconds]; oom_canary.emplace(global_context, std::move(canary_config)); } else { LOG_INFO(log, "OOM canary is disabled"); } #else if (server_settings[ServerSetting::oom_canary_enable]) LOG_WARNING(log, "OOM canary is only supported on Linux, ignoring"); #endif std::unique_ptr cancellation_task; SCOPE_EXIT({ if (cancellation_task) CancellationChecker::getInstance().terminateThread(); }); if (server_settings[ServerSetting::background_schedule_pool_size] > 1) { auto cancellation_task_holder = global_context->getSchedulePool()->createTask( StorageID::createEmpty(), "CancellationChecker", [] { CancellationChecker::getInstance().workerFunction(); } ); cancellation_task = std::make_unique(std::move(cancellation_task_holder)); (*cancellation_task)->activateAndSchedule(); } global_context->setRemoteHostFilter(config()); global_context->setHTTPHeaderFilter(config()); /// Try to increase limit on number of open files. { rlimit rlim{}; if (getrlimit(RLIMIT_NOFILE, &rlim)) throw Poco::Exception("Cannot getrlimit"); if (rlim.rlim_cur == rlim.rlim_max) { LOG_DEBUG(log, "rlimit on number of file descriptors is {}", rlim.rlim_cur); } else { rlim_t old = rlim.rlim_cur; rlim.rlim_cur = server_settings[ServerSetting::max_open_files].changed ? static_cast(server_settings[ServerSetting::max_open_files]) : rlim.rlim_max; int rc = setrlimit(RLIMIT_NOFILE, &rlim); if (rc != 0) LOG_WARNING(log, "Cannot set max number of file descriptors to {}. Try to specify max_open_files according to your system limits. error: {}", rlim.rlim_cur, errnoToString()); else LOG_DEBUG(log, "Set max number of file descriptors to {} (was {}).", rlim.rlim_cur, old); } } #if defined(RLIMIT_NPROC) /// Try to increase limit on number of threads. { rlimit rlim{}; if (getrlimit(RLIMIT_NPROC, &rlim)) throw Poco::Exception("Cannot getrlimit"); if (rlim.rlim_cur == rlim.rlim_max) { LOG_DEBUG(log, "rlimit on number of threads is {}", rlim.rlim_cur); } else { rlim_t old = rlim.rlim_cur; rlim.rlim_cur = rlim.rlim_max; int rc = setrlimit(RLIMIT_NPROC, &rlim); if (rc != 0) { LOG_WARNING(log, "Cannot set max number of threads to {}. error: {}", rlim.rlim_cur, errnoToString()); rlim.rlim_cur = old; } else { LOG_DEBUG(log, "Set max number of threads to {} (was {}).", rlim.rlim_cur, old); } } if (rlim.rlim_cur < 30000) { global_context->addOrUpdateWarningMessage( Context::WarningType::MAX_NUM_THREADS_LOWER_THAN_LIMIT, PreformattedMessage::create( "Maximum number of threads is lower than 30000. There could be problems with handling a lot of simultaneous queries.")); } } #endif static ServerErrorHandler error_handler; Poco::ErrorHandler::set(&error_handler); /// Initialize DateLUT early, to not interfere with running time of first query. LOG_DEBUG(log, "Initializing DateLUT."); DateLUT::serverTimezoneInstance(); LOG_TRACE(log, "Initialized DateLUT with time zone '{}'.", getDateLUTTimeZone(DateLUT::serverTimezoneInstance())); if (config().has("macros")) global_context->setMacros(std::make_unique(config(), "macros", log)); /** Directory with 'flags': files indicating temporary settings for the server set by system administrator. * Flags may be cleared automatically after being applied by the server. * Examples: do repair of local data; clone all replicated tables from replica. */ { auto flags_path = path / "flags/"; fs::create_directories(flags_path); global_context->setFlagsPath(flags_path); } /** Directory with user provided files that are usable by 'file' table function. */ { const auto & user_files_path_setting = server_settings[ServerSetting::user_files_path]; std::string user_files_path = user_files_path_setting.changed ? getCanonicalPath(String(user_files_path_setting.value), path_str) : String(path / "user_files/"); global_context->setUserFilesPath(user_files_path); fs::create_directories(user_files_path); } { const auto & user_scripts_path_setting = server_settings[ServerSetting::user_scripts_path]; std::string user_scripts_path = user_scripts_path_setting.changed ? getCanonicalPath(String(user_scripts_path_setting.value), path_str) : String(path / "user_scripts/"); global_context->setUserScriptsPath(user_scripts_path); } { const auto & dynamic_udf_path_setting = server_settings[ServerSetting::dynamic_user_defined_executable_functions_path]; std::string dynamic_udf_path = dynamic_udf_path_setting.changed ? getCanonicalPath(String(dynamic_udf_path_setting.value), path_str) : String(path / "dynamic_user_defined_executable_functions/"); global_context->setDynamicUserDefinedExecutableFunctionsPath(dynamic_udf_path); fs::create_directories(dynamic_udf_path); } /// top_level_domains_lists { const auto & top_level_domains_path_setting = server_settings[ServerSetting::top_level_domains_path]; std::string top_level_domains_path = top_level_domains_path_setting.changed ? getCanonicalPath(String(top_level_domains_path_setting.value), path_str) : String(path / "top_level_domains/"); TLDListsHolder::getInstance().parseConfig(fs::path(top_level_domains_path) / "", config()); } { fs::create_directories(path / "data"); fs::create_directories(path / "metadata"); /// Directory with metadata of tables, which was marked as dropped by Atomic database fs::create_directories(path / "metadata_dropped"); } if (server_settings[ServerSetting::interserver_http_port].changed && server_settings[ServerSetting::interserver_https_port].changed) throw Exception(ErrorCodes::EXCESSIVE_ELEMENT_IN_CONFIG, "Both http and https interserver ports are specified"); static const auto interserver_tags = { std::make_tuple("interserver_http_host", server_settings[ServerSetting::interserver_http_host], "interserver_http_port", server_settings[ServerSetting::interserver_http_port], "http"), std::make_tuple("interserver_https_host", server_settings[ServerSetting::interserver_https_host], "interserver_https_port", server_settings[ServerSetting::interserver_https_port], "https") }; for (auto [host_tag, host_setting, port_tag, port_setting, scheme] : interserver_tags) { if (port_setting.changed) { String this_host = host_setting.value; if (this_host.empty()) { this_host = getFQDNOrHostName(); LOG_DEBUG(log, "Configuration parameter '{}' doesn't exist or exists and empty. Will use '{}' as replica host.", host_tag, this_host); } UInt64 port = port_setting.value; if (port > 0xFFFF) throw Exception(ErrorCodes::ARGUMENT_OUT_OF_BOUND, "Out of range '{}': {}", String(port_tag), port); global_context->setInterserverIOAddress(this_host, static_cast(port)); global_context->setInterserverScheme(scheme); } } LOG_DEBUG(log, "Initializing interserver credentials."); global_context->updateInterserverCredentials(config()); std::shared_ptr stateless_worker_endpoint_ptr{nullptr}; String stateless_worker_endpoint_name; if (config().getBool("stateless_worker_server.enabled", false)) { String stateless_worker_endpoint = config().getString("stateless_worker_server.endpoint", "localhost"); size_t stateless_worker_max_threads = config().getUInt64("stateless_worker_server.max_threads", 1000); if (stateless_worker_max_threads == 0) throw Exception(ErrorCodes::INVALID_CONFIG_PARAMETER, "`stateless_worker_server.max_threads` must be at least 1"); stateless_worker_endpoint_ptr = std::make_shared( stateless_worker_max_threads, config().getUInt64("stateless_worker_server.max_free_threads", 0), config().getUInt64("stateless_worker_server.queue_size", 3000)); stateless_worker_endpoint_name = stateless_worker_endpoint_ptr->getId(stateless_worker_endpoint); global_context->getInterserverIOHandler().addEndpoint(stateless_worker_endpoint_name, stateless_worker_endpoint_ptr); LOG_DEBUG(log, "Added stateless worker endpoint '{}'.", stateless_worker_endpoint_name); } SCOPE_EXIT({ if (stateless_worker_endpoint_ptr) { /// Remove the same endpoint that was registered (the configured name may differ from "localhost"). LOG_DEBUG(log, "Shutting down stateless worker endpoint '{}'.", stateless_worker_endpoint_name); global_context->getInterserverIOHandler().removeEndpointIfExists(stateless_worker_endpoint_name); stateless_worker_endpoint_ptr->blocker.cancelForever(); stateless_worker_endpoint_ptr->shutdown(); /// Acquire the lock to wait for all in-flight requests to finish. std::lock_guard lock(stateless_worker_endpoint_ptr->rwlock); } stateless_worker_endpoint_ptr.reset(); }); #if defined(OS_LINUX) || defined(OS_DARWIN) ExchangeConnectionsPtr exchange_connections_ptr = ExchangeConnections::instance(); std::vector> exchange_servers; if (auto streaming_exchange_port = config().getUInt("distributed_query.streaming_exchange_port", 0)) { if (streaming_exchange_port > 65535) throw Exception(ErrorCodes::INVALID_CONFIG_PARAMETER, "`distributed_query.streaming_exchange_port` must be in range 1..65535, got {}", streaming_exchange_port); /// The exchange handshake is unauthenticated, so the listener is never bound to all interfaces /// implicitly: the streaming exchange is enabled only when explicit listen host(s) are given. Strings exchange_listen_hosts = DB::getMultipleValuesFromConfig(config(), "distributed_query", "streaming_exchange_listen_host"); if (exchange_listen_hosts.empty()) { LOG_ERROR(log, "`distributed_query.streaming_exchange_port` is set but no " "`distributed_query.streaming_exchange_listen_host` is configured; the streaming exchange " "server is not started. Specify a listen host to enable it."); } else { for (const auto & listen_host : exchange_listen_hosts) { try { exchange_servers.emplace_back(std::make_shared(listen_host, streaming_exchange_port, exchange_connections_ptr)); exchange_servers.back()->start(); } catch (Poco::Exception & e) { LOG_INFO(log, "Failed to start exchange server on {}:{}: {}", listen_host, streaming_exchange_port, e.displayText()); } } if (exchange_servers.empty()) throw Exception(ErrorCodes::NETWORK_ERROR, "Failed to start ExchangeServer on port {}", streaming_exchange_port); } } SCOPE_EXIT({ for (auto & exchange_server_ptr : exchange_servers) { exchange_server_ptr->stop(); exchange_server_ptr.reset(); } }); #else if (config().getUInt("distributed_query.streaming_exchange_port", 0)) throw Exception(ErrorCodes::SUPPORT_IS_DISABLED, "ExchangeServer is not supported on this platform"); #endif /// Set up caches. JemallocCacheArena::setEnabled(server_settings[ServerSetting::use_separate_cache_arena]); const size_t max_cache_size = static_cast(static_cast(physical_server_memory) * server_settings[ServerSetting::cache_size_to_ram_max_ratio]); String uncompressed_cache_policy = server_settings[ServerSetting::uncompressed_cache_policy]; size_t uncompressed_cache_size = server_settings[ServerSetting::uncompressed_cache_size]; double uncompressed_cache_size_ratio = server_settings[ServerSetting::uncompressed_cache_size_ratio]; if (uncompressed_cache_size > max_cache_size) { uncompressed_cache_size = max_cache_size; LOG_INFO(log, "Lowered uncompressed cache size to {} because the system has limited RAM", formatReadableSizeWithBinarySuffix(uncompressed_cache_size)); } global_context->setUncompressedCache(uncompressed_cache_policy, uncompressed_cache_size, uncompressed_cache_size_ratio); if (server_settings[ServerSetting::enable_gpu]) { #if USE_GPU global_context->setGPU(server_settings); #else throw Exception(ErrorCodes::SUPPORT_IS_DISABLED, "The server setting `enable_gpu` is set, but the server is built without GPU support"); #endif } String mark_cache_policy = server_settings[ServerSetting::mark_cache_policy]; size_t mark_cache_size = server_settings[ServerSetting::mark_cache_size]; double mark_cache_size_ratio = server_settings[ServerSetting::mark_cache_size_ratio]; if (mark_cache_size > max_cache_size) { mark_cache_size = max_cache_size; LOG_INFO(log, "Lowered mark cache size to {} because the system has limited RAM", formatReadableSizeWithBinarySuffix(mark_cache_size)); } global_context->setMarkCache(mark_cache_policy, mark_cache_size, mark_cache_size_ratio); String unique_key_index_cache_policy_name = server_settings[ServerSetting::unique_key_index_cache_policy]; size_t unique_key_index_cache_size = server_settings[ServerSetting::unique_key_index_cache_size_bytes]; double unique_key_index_cache_size_ratio = server_settings[ServerSetting::unique_key_index_cache_size_ratio]; if (unique_key_index_cache_size > max_cache_size) { unique_key_index_cache_size = max_cache_size; LOG_INFO(log, "Lowered UNIQUE KEY index cache size to {} because the system has limited RAM", formatReadableSizeWithBinarySuffix(unique_key_index_cache_size)); } global_context->setUniqueKeyIndexCache(unique_key_index_cache_policy_name, unique_key_index_cache_size, unique_key_index_cache_size_ratio); /// UNIQUE KEY delete-bitmap cache. Zero size disables. String unique_key_bitmap_cache_policy_name = server_settings[ServerSetting::unique_key_bitmap_cache_policy]; size_t unique_key_bitmap_cache_size = server_settings[ServerSetting::unique_key_bitmap_cache_size_bytes]; double unique_key_bitmap_cache_size_ratio = server_settings[ServerSetting::unique_key_bitmap_cache_size_ratio]; if (unique_key_bitmap_cache_size > max_cache_size) { unique_key_bitmap_cache_size = max_cache_size; LOG_INFO(log, "Lowered UNIQUE KEY delete-bitmap cache size to {} because the system has limited RAM", formatReadableSizeWithBinarySuffix(unique_key_bitmap_cache_size)); } global_context->setDeleteBitmapCache(unique_key_bitmap_cache_policy_name, unique_key_bitmap_cache_size, unique_key_bitmap_cache_size_ratio); String primary_index_cache_policy = server_settings[ServerSetting::primary_index_cache_policy]; size_t primary_index_cache_size = server_settings[ServerSetting::primary_index_cache_size]; double primary_index_cache_size_ratio = server_settings[ServerSetting::primary_index_cache_size_ratio]; if (primary_index_cache_size > max_cache_size) { primary_index_cache_size = max_cache_size; LOG_INFO(log, "Lowered primary index cache size to {} because the system has limited RAM", formatReadableSizeWithBinarySuffix(primary_index_cache_size)); } global_context->setPrimaryIndexCache(primary_index_cache_policy, primary_index_cache_size, primary_index_cache_size_ratio); String statistics_cache_policy = server_settings[ServerSetting::statistics_cache_policy]; size_t statistics_cache_size = server_settings[ServerSetting::statistics_cache_size]; double statistics_cache_size_ratio = server_settings[ServerSetting::statistics_cache_size_ratio]; if (statistics_cache_size > max_cache_size) { statistics_cache_size = max_cache_size; LOG_INFO(log, "Lowered statistics cache size to {} because the system has limited RAM", formatReadableSizeWithBinarySuffix(statistics_cache_size)); } global_context->setStatisticsCache(statistics_cache_policy, statistics_cache_size, statistics_cache_size_ratio); String columns_cache_policy = server_settings[ServerSetting::columns_cache_policy]; /// Unless configured explicitly, the columns cache is sized relative to the memory of the server. size_t columns_cache_size = config().getUInt64("columns_cache_size", getDefaultColumnsCacheSize(physical_server_memory, server_settings[ServerSetting::columns_cache_size_to_ram_ratio])); double columns_cache_size_ratio = server_settings[ServerSetting::columns_cache_size_ratio]; if (columns_cache_size > max_cache_size) { columns_cache_size = max_cache_size; LOG_INFO(log, "Lowered columns cache size to {} because the system has limited RAM", formatReadableSizeWithBinarySuffix(columns_cache_size)); } global_context->setColumnsCache(columns_cache_policy, columns_cache_size, columns_cache_size_ratio); /// The columns cache gives its memory back to the queries when the server is short of it, see /// `ColumnsCache::autoResize`: on every tick of the memory worker, and when an allocation is /// about to exceed the memory limit. if (auto columns_cache = global_context->getColumnsCache()) { columns_cache->setAutoResizeSettings( server_settings[ServerSetting::columns_cache_free_memory_ratio], server_settings[ServerSetting::columns_cache_history_window_ms]); setMemoryReleasableCache(columns_cache.get()); memory_worker.setReleasableCache(columns_cache); } String index_uncompressed_cache_policy = server_settings[ServerSetting::index_uncompressed_cache_policy]; size_t index_uncompressed_cache_size = server_settings[ServerSetting::index_uncompressed_cache_size]; double index_uncompressed_cache_size_ratio = server_settings[ServerSetting::index_uncompressed_cache_size_ratio]; if (index_uncompressed_cache_size > max_cache_size) { index_uncompressed_cache_size = max_cache_size; LOG_INFO(log, "Lowered index uncompressed cache size to {} because the system has limited RAM", formatReadableSizeWithBinarySuffix(index_uncompressed_cache_size)); } global_context->setIndexUncompressedCache(index_uncompressed_cache_policy, index_uncompressed_cache_size, index_uncompressed_cache_size_ratio); String index_mark_cache_policy = server_settings[ServerSetting::index_mark_cache_policy]; size_t index_mark_cache_size = server_settings[ServerSetting::index_mark_cache_size]; double index_mark_cache_size_ratio = server_settings[ServerSetting::index_mark_cache_size_ratio]; if (index_mark_cache_size > max_cache_size) { index_mark_cache_size = max_cache_size; LOG_INFO(log, "Lowered index mark cache size to {} because the system has limited RAM", formatReadableSizeWithBinarySuffix(index_mark_cache_size)); } global_context->setIndexMarkCache(index_mark_cache_policy, index_mark_cache_size, index_mark_cache_size_ratio); String vector_similarity_index_cache_policy = server_settings[ServerSetting::vector_similarity_index_cache_policy]; size_t vector_similarity_index_cache_size = server_settings[ServerSetting::vector_similarity_index_cache_size]; size_t vector_similarity_index_cache_max_entries = server_settings[ServerSetting::vector_similarity_index_cache_max_entries]; double vector_similarity_index_cache_size_ratio = server_settings[ServerSetting::vector_similarity_index_cache_size_ratio]; if (vector_similarity_index_cache_size > max_cache_size) { vector_similarity_index_cache_size = max_cache_size; LOG_INFO(log, "Lowered vector similarity index cache size to {} because the system has limited RAM", formatReadableSizeWithBinarySuffix(vector_similarity_index_cache_size)); } global_context->setVectorSimilarityIndexCache(vector_similarity_index_cache_policy, vector_similarity_index_cache_size, vector_similarity_index_cache_max_entries, vector_similarity_index_cache_size_ratio); String text_index_tokens_cache_policy = server_settings[ServerSetting::text_index_tokens_cache_policy]; size_t text_index_tokens_cache_size = server_settings[ServerSetting::text_index_tokens_cache_size]; size_t text_index_tokens_cache_max_count = server_settings[ServerSetting::text_index_tokens_cache_max_entries]; double text_index_tokens_cache_size_ratio = server_settings[ServerSetting::text_index_tokens_cache_size_ratio]; if (text_index_tokens_cache_size > max_cache_size) { text_index_tokens_cache_size = max_cache_size; LOG_INFO(log, "Lowered text index tokens cache size to {} because the system has limited RAM", formatReadableSizeWithBinarySuffix(text_index_tokens_cache_size)); } global_context->setTextIndexTokensCache(text_index_tokens_cache_policy, text_index_tokens_cache_size, text_index_tokens_cache_max_count, text_index_tokens_cache_size_ratio); String text_index_header_cache_policy = server_settings[ServerSetting::text_index_header_cache_policy]; size_t text_index_header_cache_size = server_settings[ServerSetting::text_index_header_cache_size]; size_t text_index_header_cache_max_count = server_settings[ServerSetting::text_index_header_cache_max_entries]; double text_index_header_cache_size_ratio = server_settings[ServerSetting::text_index_header_cache_size_ratio]; if (text_index_header_cache_size > max_cache_size) { text_index_header_cache_size = max_cache_size; LOG_INFO(log, "Lowered text index header cache size to {} because the system has limited RAM", formatReadableSizeWithBinarySuffix(text_index_header_cache_size)); } global_context->setTextIndexHeaderCache(text_index_header_cache_policy, text_index_header_cache_size, text_index_header_cache_max_count, text_index_header_cache_size_ratio); String text_index_postings_cache_policy = server_settings[ServerSetting::text_index_postings_cache_policy]; size_t text_index_postings_cache_size = server_settings[ServerSetting::text_index_postings_cache_size]; size_t text_index_postings_cache_max_count = server_settings[ServerSetting::text_index_postings_cache_max_entries]; double text_index_postings_cache_size_ratio = server_settings[ServerSetting::text_index_postings_cache_size_ratio]; if (text_index_postings_cache_size > max_cache_size) { text_index_postings_cache_size = max_cache_size; LOG_INFO(log, "Lowered text index posting list cache size to {} because the system has limited RAM", formatReadableSizeWithBinarySuffix(text_index_postings_cache_size)); } global_context->setTextIndexPostingsCache(text_index_postings_cache_policy, text_index_postings_cache_size, text_index_postings_cache_max_count, text_index_postings_cache_size_ratio); size_t mmap_cache_size = server_settings[ServerSetting::mmap_cache_size]; if (mmap_cache_size > max_cache_size) { mmap_cache_size = max_cache_size; LOG_INFO(log, "Lowered mmap file cache size to {} because the system has limited RAM", formatReadableSizeWithBinarySuffix(mmap_cache_size)); } global_context->setMMappedFileCache(mmap_cache_size); #if USE_AVRO String iceberg_metadata_files_cache_policy = server_settings[ServerSetting::iceberg_metadata_files_cache_policy]; size_t iceberg_metadata_files_cache_size = server_settings[ServerSetting::iceberg_metadata_files_cache_size]; size_t iceberg_metadata_files_cache_max_entries = server_settings[ServerSetting::iceberg_metadata_files_cache_max_entries]; double iceberg_metadata_files_cache_size_ratio = server_settings[ServerSetting::iceberg_metadata_files_cache_size_ratio]; if (iceberg_metadata_files_cache_size > max_cache_size) { iceberg_metadata_files_cache_size = max_cache_size; LOG_INFO(log, "Lowered Iceberg metadata cache size to {} because the system has limited RAM", formatReadableSizeWithBinarySuffix(iceberg_metadata_files_cache_size)); } global_context->setIcebergMetadataFilesCache(iceberg_metadata_files_cache_policy, iceberg_metadata_files_cache_size, iceberg_metadata_files_cache_max_entries, iceberg_metadata_files_cache_size_ratio); String paimon_metadata_files_cache_policy = server_settings[ServerSetting::paimon_metadata_files_cache_policy]; size_t paimon_metadata_files_cache_size = server_settings[ServerSetting::paimon_metadata_files_cache_size]; size_t paimon_metadata_files_cache_max_entries = server_settings[ServerSetting::paimon_metadata_files_cache_max_entries]; double paimon_metadata_files_cache_size_ratio = server_settings[ServerSetting::paimon_metadata_files_cache_size_ratio]; if (paimon_metadata_files_cache_size > max_cache_size) { paimon_metadata_files_cache_size = max_cache_size; LOG_INFO(log, "Lowered Paimon metadata cache size to {} because the system has limited RAM", formatReadableSizeWithBinarySuffix(paimon_metadata_files_cache_size)); } global_context->setPaimonMetadataFilesCache(paimon_metadata_files_cache_policy, paimon_metadata_files_cache_size, paimon_metadata_files_cache_max_entries, paimon_metadata_files_cache_size_ratio); #endif #if USE_PARQUET String parquet_metadata_cache_policy = server_settings[ServerSetting::parquet_metadata_cache_policy]; size_t parquet_metadata_cache_size = server_settings[ServerSetting::parquet_metadata_cache_size]; size_t parquet_metadata_cache_max_entries = server_settings[ServerSetting::parquet_metadata_cache_max_entries]; double parquet_metadata_cache_size_ratio = server_settings[ServerSetting::parquet_metadata_cache_size_ratio]; if (parquet_metadata_cache_size > max_cache_size) { parquet_metadata_cache_size = max_cache_size; LOG_INFO(log, "Lowered Parquet metadata cache size to {} because the system has limited RAM", formatReadableSizeWithBinarySuffix(parquet_metadata_cache_size)); } global_context->setParquetMetadataCache(parquet_metadata_cache_policy, parquet_metadata_cache_size, parquet_metadata_cache_max_entries, parquet_metadata_cache_size_ratio); #endif Names allowed_disks_table_engines; splitInto<','>(allowed_disks_table_engines, server_settings[ServerSetting::allowed_disks_for_table_engines].value); global_context->setAllowedDisksForTableEngines(std::unordered_set(allowed_disks_table_engines.begin(), allowed_disks_table_engines.end())); String query_condition_cache_policy = server_settings[ServerSetting::query_condition_cache_policy]; size_t query_condition_cache_size = server_settings[ServerSetting::query_condition_cache_size]; double query_condition_cache_size_ratio = server_settings[ServerSetting::query_condition_cache_size_ratio]; if (query_condition_cache_size > max_cache_size) { query_condition_cache_size = max_cache_size; LOG_INFO(log, "Lowered query condition cache size to {} because the system has limited RAM", formatReadableSizeWithBinarySuffix(query_condition_cache_size)); } global_context->setQueryConditionCache(query_condition_cache_policy, query_condition_cache_size, query_condition_cache_size_ratio); String encryption_header_cache_policy = server_settings[ServerSetting::encryption_header_cache_policy]; size_t encryption_header_cache_size = server_settings[ServerSetting::encryption_header_cache_size]; double encryption_header_cache_size_ratio = server_settings[ServerSetting::encryption_header_cache_size_ratio]; if (encryption_header_cache_size > max_cache_size) { encryption_header_cache_size = max_cache_size; LOG_INFO(log, "Lowered encryption header cache size to {} because the system has limited RAM", formatReadableSizeWithBinarySuffix(encryption_header_cache_size)); } global_context->setEncryptionHeaderCache(encryption_header_cache_policy, encryption_header_cache_size, encryption_header_cache_size_ratio); size_t query_result_cache_max_size_in_bytes = server_settings[ServerSetting::query_cache_max_size_in_bytes]; size_t query_result_cache_max_entries = server_settings[ServerSetting::query_cache_max_entries]; size_t query_result_cache_max_entry_size_in_bytes = server_settings[ServerSetting::query_cache_max_entry_size_in_bytes]; size_t query_result_cache_max_entry_size_in_rows = server_settings[ServerSetting::query_cache_max_entry_size_in_rows]; if (query_result_cache_max_size_in_bytes > max_cache_size) { query_result_cache_max_size_in_bytes = max_cache_size; LOG_INFO(log, "Lowered query result cache size to {} because the system has limited RAM", formatReadableSizeWithBinarySuffix(query_result_cache_max_size_in_bytes)); } global_context->setQueryResultCache(query_result_cache_max_size_in_bytes, query_result_cache_max_entries, query_result_cache_max_entry_size_in_bytes, query_result_cache_max_entry_size_in_rows); #if USE_EMBEDDED_COMPILER size_t compiled_expression_cache_max_size_in_bytes = server_settings[ServerSetting::compiled_expression_cache_size]; size_t compiled_expression_cache_max_elements = server_settings[ServerSetting::compiled_expression_cache_elements_size]; CompiledExpressionCacheFactory::instance().init(compiled_expression_cache_max_size_in_bytes, compiled_expression_cache_max_elements); #endif size_t point_in_polygon_cache_size = server_settings[ServerSetting::point_in_polygon_cache_size]; if (point_in_polygon_cache_size > max_cache_size) { point_in_polygon_cache_size = max_cache_size; LOG_INFO(log, "Lowered point in polygon cache size to {} because the system has limited RAM", formatReadableSizeWithBinarySuffix(point_in_polygon_cache_size)); } setPointInPolygonCacheMaxSizeInBytes(point_in_polygon_cache_size); NamedCollectionFactory::instance().loadIfNot(); SQLDefinedHandlersFactory::instance().loadIfNot(); FileCacheFactory::instance().loadDefaultCaches(config(), global_context); /// Initialize main config reloader. std::string include_from_path = server_settings[ServerSetting::include_from]; if (config().has("query_masking_rules")) { SensitiveDataMasker::setInstance(std::make_unique(config(), "query_masking_rules")); } std::optional cgroups_memory_usage_observer; try { const auto & wait_time = server_settings[ServerSetting::cgroups_memory_usage_observer_wait_time]; if (wait_time != 0) cgroups_memory_usage_observer.emplace(std::chrono::seconds(wait_time)); } catch (Exception &) { tryLogCurrentException(log, "Disabling cgroup memory observer because of an error during initialization"); } /// TLS certificates, keys and CA certificates are reloaded by CertificateReloader when these files change. std::vector extra_paths = {include_from_path}; auto watch_path = [&](const std::string & file_path) { if (!file_path.empty()) extra_paths.emplace_back(file_path); }; watch_path(server_settings[ServerSetting::openssl_server_certificate_file]); watch_path(server_settings[ServerSetting::openssl_server_private_key_file]); watch_path(server_settings[ServerSetting::openssl_server_ca_config]); watch_path(server_settings[ServerSetting::openssl_client_certificate_file]); watch_path(server_settings[ServerSetting::openssl_client_private_key_file]); watch_path(server_settings[ServerSetting::openssl_client_ca_config]); Poco::Util::AbstractConfiguration::Keys protocols; config().keys("protocols", protocols); for (const auto & protocol : protocols) for (const auto * key : {"certificateFile", "privateKeyFile", "caConfig"}) watch_path(config().getString("protocols." + protocol + "." + key, "")); DNSResolver::instance().setFilterSettings(server_settings[ServerSetting::dns_allow_resolve_names_to_ipv4], server_settings[ServerSetting::dns_allow_resolve_names_to_ipv6]); /// DNSCacheUpdater uses BackgroundSchedulePool which lives in shared context /// and thus this object must be created after the SCOPE_EXIT object where shared /// context is destroyed. /// In addition this object has to be created before any scenario that requires /// DNS resolution (e.g. the loading of tables and clusters config). std::unique_ptr dns_cache_updater; if (server_settings[ServerSetting::disable_internal_dns_cache]) { /// Disable DNS caching at all DNSResolver::instance().setDisableCacheFlag(); LOG_DEBUG(log, "DNS caching disabled"); } else { DNSResolver::instance().setCacheMaxEntries(server_settings[ServerSetting::dns_cache_max_entries]); /// Prime the local host name / addresses cache before the updater starts, /// so that the first DNSCacheUpdater tick on a healthy server does not /// see an empty cache and spuriously report "IPs of host name X have been changed". DNSResolver::instance().updateHostNameAndAddresses(); /// Initialize a watcher periodically updating DNS cache dns_cache_updater = std::make_unique( global_context, server_settings[ServerSetting::dns_cache_update_period], server_settings[ServerSetting::dns_max_consecutive_failures]); dns_cache_updater->start(); } /// Whether `--skip_check_for_incorrect_settings=1` was passed on the *command line*, captured /// independently of any config file layer. Reusing the layered `config()` here would conflate /// the command-line value with whatever the *previously loaded* file set: a config that once /// contained `1` would /// leave that flag live in `config()`, so a reload that *removes* the flag while adding an /// unknown key would still be skipped — diverging from a fresh startup with the same file, /// which would reject it. Re-parsing `argv()` with `argsToConfig` (exactly as `BaseDaemon` /// populates the command-line layer) reproduces the command-line-only view; the priority is /// irrelevant because the throwaway config has a single layer. bool command_line_skip_check = false; { Poco::AutoPtr command_line_config(new Poco::Util::LayeredConfiguration); argsToConfig(argv(), *command_line_config, PRIO_DEFAULT); command_line_skip_check = command_line_config->getBool("skip_check_for_incorrect_settings", false); } auto main_config_reloader = std::make_unique( config_path, extra_paths, server_settings[ServerSetting::path], std::move(main_config_zk_node_cache), main_config_zk_changed_event, [&](ConfigurationPtr loaded_config, bool initial_loading) { /// Validate the new config BEFORE installing it into the global layered config, so a /// failed reload does not leave `config()` partially mutated with unvalidated changes. /// The escape hatch is honored when it is either passed on the command line (persisted /// across reloads) or present in the *new* file being validated — never inherited from /// the previously loaded file layer (see `command_line_skip_check` above). This matches /// startup: validating the same file fresh must reach the same accept/reject decision. const bool skip_check = command_line_skip_check || loaded_config->getBool("skip_check_for_incorrect_settings", false); if (!skip_check) Settings::checkNoSettingNamesAtTopLevel(*loaded_config, config_path); /// Same as on initial load: validate the reloaded XML config rather than the layered /// view, so CLI-injected and Poco-internal top-level keys do not need an allowlist. ServerSettings::checkUnknownSettings(*loaded_config, config_path, skip_check); /// Fail closed on a legacy insert_deduplication_version, or on a memory-pressure threshold /// triple that is out of range or out of order, arriving via a runtime reload. /// Validate the incoming config BEFORE config().replace below: validating after would mutate /// the live config even for a rejected reload (ConfigReloader has no rollback hook), leaving /// system.server_settings reporting an unsupported value. Reject first, then replace. { ServerSettings incoming_server_settings; incoming_server_settings.loadSettingsFromConfig(*loaded_config); validate_insert_deduplication_version(incoming_server_settings); validateMemoryPressureThresholds( incoming_server_settings[ServerSetting::reader_executor_memory_pressure_elevated_level_pct], incoming_server_settings[ServerSetting::reader_executor_memory_pressure_high_level_pct], incoming_server_settings[ServerSetting::reader_executor_memory_pressure_critical_level_pct]); } /// The Prometheus check below, the reloaders and `latest_config` keep the configuration they are given, so it must have /// the command line on top of the file and, like `config()`, resolve `${application.*}` and `${system.*}` references. Poco::AutoPtr file_and_command_line(new Poco::Util::LayeredConfiguration); file_and_command_line->add(loaded_config, PRIO_DEFAULT); argsToConfig(argv(), *file_and_command_line, PRIO_APPLICATION - 100); file_and_command_line->add(application_properties, PRIO_APPLICATION, /* shared */ true); file_and_command_line->add(new Poco::Util::SystemConfiguration, PRIO_SYSTEM, /* shared */ false); const ConfigurationPtr reload_config = file_and_command_line; /// Fail closed on a Prometheus constant label that collides with a label an endpoint writes /// itself. `asynchronous_metrics_key_values_mode` takes part in this check, because it decides /// whether the key of a key-value asynchronous metric is written as a label (`device="sda"`), /// so the same set of constant labels can be unambiguous under one form and not under another. /// Validate the incoming config BEFORE config().replace below, for the same reason as above: /// the form is read from the live configuration on every update of the asynchronous metrics, /// so validating afterwards would leave a rejected reload publishing the new form while the /// endpoints keep serving with the labels of the old one. validatePrometheusConstantLabels( *reload_config, servedHTTPHandlersKeys(*reload_config), hasPrometheusListener(*reload_config)); config().replace("default", loaded_config, PRIO_DEFAULT, true); ServerSettings new_server_settings; new_server_settings.loadSettingsFromConfig(config()); /// The check above sees the incoming file alone. A triple split across the file and the /// command-line layer can have each source valid and the merge out of order, so check the /// merged view too - here, before the first live setting below is touched. validateMemoryPressureThresholds( new_server_settings[ServerSetting::reader_executor_memory_pressure_elevated_level_pct], new_server_settings[ServerSetting::reader_executor_memory_pressure_high_level_pct], new_server_settings[ServerSetting::reader_executor_memory_pressure_critical_level_pct]); DB::abort_on_logical_error.store(new_server_settings[ServerSetting::abort_on_logical_error], std::memory_order_relaxed); ConnectionRegistry::instance().setEnabled(new_server_settings[ServerSetting::collect_connection_metrics]); /// The seccomp filter cannot be changed once installed; `system.server_settings` keeps showing the mode in force. if (const auto requested_seccomp_mode = getRequestedSeccompMode(); requested_seccomp_mode && *requested_seccomp_mode != new_server_settings[ServerSetting::seccomp].value) LOG_WARNING( log, "The `seccomp` server setting was changed from `{}` to `{}` in the configuration, but it takes effect only " "after a restart: the seccomp policy of a running process cannot be changed", SettingFieldSeccompMode(*requested_seccomp_mode).toString(), new_server_settings[ServerSetting::seccomp].toString()); size_t max_server_memory_usage = new_server_settings[ServerSetting::max_server_memory_usage]; const double max_server_memory_usage_to_ram_ratio = new_server_settings[ServerSetting::max_server_memory_usage_to_ram_ratio]; const size_t current_physical_server_memory = getMemoryAmount(); /// With cgroups, the amount of memory available to the server can be changed dynamically. const size_t default_max_server_memory_usage = static_cast(static_cast(current_physical_server_memory) * max_server_memory_usage_to_ram_ratio); if (max_server_memory_usage == 0) { max_server_memory_usage = default_max_server_memory_usage; LOG_INFO(log, "Changed setting 'max_server_memory_usage' to {}" " ({} available memory * {:.2f} max_server_memory_usage_to_ram_ratio)", formatReadableSizeWithBinarySuffix(max_server_memory_usage), formatReadableSizeWithBinarySuffix(current_physical_server_memory), max_server_memory_usage_to_ram_ratio); } else if (max_server_memory_usage > default_max_server_memory_usage) { max_server_memory_usage = default_max_server_memory_usage; LOG_INFO(log, "Lowered setting 'max_server_memory_usage' to {}" " because the system has too little memory. The new value was" " calculated as {} available memory * {:.2f} max_server_memory_usage_to_ram_ratio", formatReadableSizeWithBinarySuffix(max_server_memory_usage), formatReadableSizeWithBinarySuffix(current_physical_server_memory), max_server_memory_usage_to_ram_ratio); } total_memory_tracker.setDescription("(total)"); total_memory_tracker.setMetric(CurrentMetrics::MemoryTracking); /// Inform `MemoryWorker` of the configured ceiling and the ratio so its dynamic /// adjustment (which only sees `MemAvailable` or cgroup memory) cannot exceed /// the explicit `max_server_memory_usage`, and so a config-reload change to /// `max_server_memory_usage_to_ram_ratio` takes effect on the next worker tick. /// `setDynamicHardLimitSettings` also installs `max_server_memory_usage` as the /// new hard limit atomically with the settings update; doing it outside that call /// would re-open a reload race against the worker tick. /// A zero ratio disables the runtime adjustment, keeping only the static cap; /// `memory_worker_dynamic_hard_limit = 0` requests exactly that. memory_worker.setDynamicHardLimitSettings( static_cast(max_server_memory_usage), new_server_settings[ServerSetting::memory_worker_dynamic_hard_limit] ? max_server_memory_usage_to_ram_ratio : 0.0); CurrentMemoryTracker::setMinAllocationSizeBytesToThrow( new_server_settings[ServerSetting::min_allocation_size_to_throw_on_memory_limit]); MemoryTracker::setMinAllocationSizeToLogStackTrace( new_server_settings[ServerSetting::min_allocation_size_to_log_stack_trace]); per_cpu_memory.setBudgetCapacity(new_server_settings[ServerSetting::max_per_cpu_untracked_memory]); per_cpu_memory.setThreadBuffer(new_server_settings[ServerSetting::per_cpu_untracked_memory_thread_buffer]); size_t merges_mutations_memory_usage_soft_limit = new_server_settings[ServerSetting::merges_mutations_memory_usage_soft_limit]; const double merges_mutations_memory_usage_to_ram_ratio = new_server_settings[ServerSetting::merges_mutations_memory_usage_to_ram_ratio]; size_t default_merges_mutations_server_memory_usage = static_cast(static_cast(current_physical_server_memory) * merges_mutations_memory_usage_to_ram_ratio); if (merges_mutations_memory_usage_soft_limit == 0) { merges_mutations_memory_usage_soft_limit = default_merges_mutations_server_memory_usage; LOG_INFO(log, "Setting merges_mutations_memory_usage_soft_limit was set to {}" " ({} available * {:.2f} merges_mutations_memory_usage_to_ram_ratio)", formatReadableSizeWithBinarySuffix(merges_mutations_memory_usage_soft_limit), formatReadableSizeWithBinarySuffix(current_physical_server_memory), merges_mutations_memory_usage_to_ram_ratio); } else if (merges_mutations_memory_usage_soft_limit > default_merges_mutations_server_memory_usage) { merges_mutations_memory_usage_soft_limit = default_merges_mutations_server_memory_usage; LOG_WARNING(log, "Setting merges_mutations_memory_usage_soft_limit was set to {}" " ({} available * {:.2f} merges_mutations_memory_usage_to_ram_ratio)", formatReadableSizeWithBinarySuffix(merges_mutations_memory_usage_soft_limit), formatReadableSizeWithBinarySuffix(current_physical_server_memory), merges_mutations_memory_usage_to_ram_ratio); } LOG_INFO(log, "Merges and mutations memory limit is set to {}", formatReadableSizeWithBinarySuffix(merges_mutations_memory_usage_soft_limit)); background_memory_tracker.setSoftLimit(merges_mutations_memory_usage_soft_limit); background_memory_tracker.setDescription("(background)"); background_memory_tracker.setMetric(CurrentMetrics::MergesMutationsMemoryTracking); auto * global_overcommit_tracker = global_context->getGlobalOvercommitTracker(); total_memory_tracker.setOvercommitTracker(global_overcommit_tracker); // FIXME logging-related things need synchronization -- see the 'Logger * log' saved // in a lot of places. For now, disable updating log configuration without server restart. //setTextLog(global_context->getTextLog()); updateLevels(config(), logger()); /// File-only, so `remote_servers` ignores the command line: `Cluster` takes the server's own `tcp_port` from this configuration /// to find its local replicas, and with `-- --tcp_port` the replicas on the file's port would become remote. global_context->setClustersConfig(loaded_config, has_zookeeper); global_context->setMacros(std::make_unique(config(), "macros", log)); global_context->setExternalAuthenticatorsConfig(config()); global_context->setDashboardsConfig(config()); if (global_context->isServerCompletelyStarted()) { /// It does not make sense to reload anything before server has started. /// Moreover, it may break initialization order. global_context->loadOrReloadDictionaries(config()); global_context->loadUserDefinedExecutableFunctionDrivers(config()); global_context->loadOrReloadUserDefinedExecutableFunctions(config()); } global_context->setRemoteHostFilter(config()); global_context->setHTTPHeaderFilter(config()); global_context->setMaxTableSizeToDrop(new_server_settings[ServerSetting::max_table_size_to_drop]); global_context->setMaxPartitionSizeToDrop(new_server_settings[ServerSetting::max_partition_size_to_drop]); global_context->setMaxNamedCollectionNumToWarn(new_server_settings[ServerSetting::max_named_collection_num_to_warn]); global_context->setMaxTableNumToWarn(new_server_settings[ServerSetting::max_table_num_to_warn]); global_context->setMaxViewNumToWarn(new_server_settings[ServerSetting::max_view_num_to_warn]); global_context->setMaxDictionaryNumToWarn(new_server_settings[ServerSetting::max_dictionary_num_to_warn]); global_context->setMaxDatabaseNumToWarn(new_server_settings[ServerSetting::max_database_num_to_warn]); global_context->setMaxPartNumToWarn(new_server_settings[ServerSetting::max_part_num_to_warn]); global_context->setMaxNamedCollectionNumToThrow(new_server_settings[ServerSetting::max_named_collection_num_to_throw]); global_context->setMaxTableNumToThrow(new_server_settings[ServerSetting::max_table_num_to_throw]); global_context->setMaxReplicatedTableNumToThrow(new_server_settings[ServerSetting::max_replicated_table_num_to_throw]); global_context->setMaxViewNumToThrow(new_server_settings[ServerSetting::max_view_num_to_throw]); global_context->setMaxDictionaryNumToThrow(new_server_settings[ServerSetting::max_dictionary_num_to_throw]); global_context->setMaxDatabaseNumToThrow(new_server_settings[ServerSetting::max_database_num_to_throw]); global_context->setMaxPendingMutationsToWarn(new_server_settings[ServerSetting::max_pending_mutations_to_warn]); global_context->setMaxPendingMutationsExecutionTimeToWarn(new_server_settings[ServerSetting::max_pending_mutations_execution_time_to_warn]); global_context->getAccessControl().setAllowTierSettings(new_server_settings[ServerSetting::allow_feature_tier]); global_context->setUsersToIgnoreEarlyMemoryLimitCheck(new_server_settings[ServerSetting::users_to_ignore_early_memory_limit_check]); global_context->allowSystemAllocateMemory(config().getBool("allow_system_allocate_memory", false)); global_context->setS3QueueDisableStreaming(new_server_settings[ServerSetting::s3queue_disable_streaming]); global_context->setReadThroughDistributedCache(new_server_settings[ServerSetting::enable_read_through_distributed_cache]); global_context->setWriteThroughDistributedCache(new_server_settings[ServerSetting::enable_write_through_distributed_cache]); global_context->setOSCPUOverloadSettings(static_cast(new_server_settings[ServerSetting::min_os_cpu_wait_time_ratio_to_drop_connection]), static_cast(new_server_settings[ServerSetting::max_os_cpu_wait_time_ratio_to_drop_connection])); size_t remote_read_bandwidth = new_server_settings[ServerSetting::max_remote_read_network_bandwidth_for_server]; size_t remote_write_bandwidth = new_server_settings[ServerSetting::max_remote_write_network_bandwidth_for_server]; global_context->reloadRemoteThrottlerConfig(remote_read_bandwidth,remote_write_bandwidth); LOG_INFO(log, "Setting max_remote_read_network_bandwidth_for_server was set to {}", remote_read_bandwidth); LOG_INFO(log, "Setting max_remote_write_network_bandwidth_for_server was set to {}", remote_write_bandwidth); size_t local_read_bandwidth = new_server_settings[ServerSetting::max_local_read_bandwidth_for_server]; size_t local_write_bandwidth = new_server_settings[ServerSetting::max_local_write_bandwidth_for_server]; global_context->reloadLocalThrottlerConfig(local_read_bandwidth,local_write_bandwidth); LOG_INFO(log, "Setting max_local_read_bandwidth_for_server was set to {}", local_read_bandwidth); LOG_INFO(log, "Setting max_local_write_bandwidth_for_server was set to {}", local_write_bandwidth); size_t max_remote_read_connections = new_server_settings[ServerSetting::max_remote_read_connections]; global_context->reloadLongConnectionLimitConfig(max_remote_read_connections); LOG_INFO(log, "Setting max_remote_read_connections was set to {}", max_remote_read_connections); #if ENABLE_DISTRIBUTED_CACHE for (const auto & distr_cache_instance : distr_cache_instances) distr_cache_instance->updateConfig(config()); #endif /// Only for system.server_settings global_context->setConfigReloaderInterval(new_server_settings[ServerSetting::config_reload_interval_ms]); auto [concurrent_threads_soft_limit, concurrency_control_scheduler] = global_context->setConcurrentThreadsSoftLimit( new_server_settings[ServerSetting::concurrent_threads_soft_limit_num], new_server_settings[ServerSetting::concurrent_threads_soft_limit_ratio_to_cores], new_server_settings[ServerSetting::concurrent_threads_scheduler], new_server_settings[ServerSetting::concurrent_threads_lazy_allocation]); LOG_INFO(log, "ConcurrencyControl limit is set to {} CPU slots with '{}' scheduler", concurrent_threads_soft_limit == UnlimitedSlots ? std::string("UNLIMITED") : std::to_string(concurrent_threads_soft_limit), concurrency_control_scheduler); global_context->getProcessList().setMaxSize(new_server_settings[ServerSetting::max_concurrent_queries]); global_context->getProcessList().setMaxInsertQueriesAmount(new_server_settings[ServerSetting::max_concurrent_insert_queries]); global_context->getProcessList().setMaxSelectQueriesAmount(new_server_settings[ServerSetting::max_concurrent_select_queries]); global_context->getProcessList().setMaxWaitingQueriesAmount(new_server_settings[ServerSetting::max_waiting_queries]); if (config().has("keeper_server")) { global_context->updateKeeperConfiguration(config()); } /// Reload the number of threads for global pools. /// Note: If you specified it in the top level config (not it config of default profile) /// then ClickHouse will use it exactly. /// This is done for backward compatibility. if (global_context->areBackgroundExecutorsInitialized()) { const auto & new_pool_size = new_server_settings[ServerSetting::background_pool_size]; const auto & new_ratio = new_server_settings[ServerSetting::background_merges_mutations_concurrency_ratio]; global_context->getMergeMutateExecutor()->increaseThreadsAndMaxTasksCount(new_pool_size, static_cast(static_cast(new_pool_size) * static_cast(new_ratio))); global_context->getMergeMutateExecutor()->updateSchedulingPolicy(new_server_settings[ServerSetting::background_merges_mutations_scheduling_policy].toString()); } if (global_context->areBackgroundExecutorsInitialized()) { const auto & new_pool_size = new_server_settings[ServerSetting::background_move_pool_size]; global_context->getMovesExecutor()->increaseThreadsAndMaxTasksCount(new_pool_size, new_pool_size); } if (global_context->areBackgroundExecutorsInitialized()) { const auto & new_pool_size = new_server_settings[ServerSetting::background_fetches_pool_size]; global_context->getFetchesExecutor()->increaseThreadsAndMaxTasksCount(new_pool_size, new_pool_size); } if (global_context->areBackgroundExecutorsInitialized()) { const auto & new_pool_size = new_server_settings[ServerSetting::background_common_pool_size]; global_context->getCommonExecutor()->increaseThreadsAndMaxTasksCount(new_pool_size, new_pool_size); } global_context->getBufferFlushSchedulePool()->increaseThreadsCount(new_server_settings[ServerSetting::background_buffer_flush_schedule_pool_size]); global_context->getSchedulePool()->increaseThreadsCount(new_server_settings[ServerSetting::background_schedule_pool_size]); global_context->getMessageBrokerSchedulePool()->increaseThreadsCount(new_server_settings[ServerSetting::background_message_broker_schedule_pool_size]); global_context->getDistributedSchedulePool()->increaseThreadsCount(new_server_settings[ServerSetting::background_distributed_schedule_pool_size]); global_context->getStreamingSchedulePool()->increaseThreadsCount(new_server_settings[ServerSetting::background_streaming_schedule_pool_size]); global_context->getAsyncLoader().setMaxThreads(TablesLoaderForegroundPoolId, new_server_settings[ServerSetting::tables_loader_foreground_pool_size]); global_context->getAsyncLoader().setMaxThreads(TablesLoaderBackgroundLoadPoolId, new_server_settings[ServerSetting::tables_loader_background_pool_size]); global_context->getAsyncLoader().setMaxThreads(TablesLoaderBackgroundStartupPoolId, new_server_settings[ServerSetting::tables_loader_background_pool_size]); getIOThreadPool().reloadConfiguration( new_server_settings[ServerSetting::max_io_thread_pool_size], new_server_settings[ServerSetting::max_io_thread_pool_free_size], new_server_settings[ServerSetting::io_thread_pool_queue_size]); getBackupsIOThreadPool().reloadConfiguration( new_server_settings[ServerSetting::max_backups_io_thread_pool_size], new_server_settings[ServerSetting::max_backups_io_thread_pool_free_size], new_server_settings[ServerSetting::backups_io_thread_pool_queue_size]); getFetchPartitionThreadPool().reloadConfiguration( new_server_settings[ServerSetting::max_fetch_partition_thread_pool_size], 0, // FETCH PARTITION is relatively rare, no need to keep threads new_server_settings[ServerSetting::max_fetch_partition_thread_pool_size]); getActivePartsLoadingThreadPool().reloadConfiguration( new_server_settings[ServerSetting::max_active_parts_loading_thread_pool_size], 0, // We don't need any threads once all the parts will be loaded new_server_settings[ServerSetting::max_active_parts_loading_thread_pool_size]); getOutdatedPartsLoadingThreadPool().reloadConfiguration( new_server_settings[ServerSetting::max_outdated_parts_loading_thread_pool_size], 0, // We don't need any threads once all the parts will be loaded new_server_settings[ServerSetting::max_outdated_parts_loading_thread_pool_size]); /// It could grow if we need to synchronously wait until all the data parts will be loaded. getOutdatedPartsLoadingThreadPool().setMaxTurboThreads( new_server_settings[ServerSetting::max_active_parts_loading_thread_pool_size] ); getPartsCleaningThreadPool().reloadConfiguration( new_server_settings[ServerSetting::max_parts_cleaning_thread_pool_size], 0, // We don't need any threads one all the parts will be deleted new_server_settings[ServerSetting::max_parts_cleaning_thread_pool_size]); getMergeTreePrefixesDeserializationThreadPool().reloadConfiguration( new_server_settings[ServerSetting::max_prefixes_deserialization_thread_pool_size], new_server_settings[ServerSetting::max_prefixes_deserialization_thread_pool_free_size], new_server_settings[ServerSetting::prefixes_deserialization_thread_pool_thread_pool_queue_size]); getFormatParsingThreadPool().reloadConfiguration( new_server_settings[ServerSetting::max_format_parsing_thread_pool_size], new_server_settings[ServerSetting::max_format_parsing_thread_pool_free_size], new_server_settings[ServerSetting::format_parsing_thread_pool_queue_size]); getAIRequestThreadPool().reloadConfiguration( new_server_settings[ServerSetting::ai_function_max_thread_pool_size], new_server_settings[ServerSetting::ai_function_max_thread_pool_free_size], new_server_settings[ServerSetting::ai_function_thread_pool_queue_size]); global_context->setMergeWorkload(new_server_settings[ServerSetting::merge_workload]); global_context->setMutationWorkload(new_server_settings[ServerSetting::mutation_workload]); global_context->setThrowOnUnknownWorkload(new_server_settings[ServerSetting::throw_on_unknown_workload]); global_context->setCPUSlotPreemption( new_server_settings[ServerSetting::cpu_slot_preemption], new_server_settings[ServerSetting::cpu_slot_quantum_ns], new_server_settings[ServerSetting::cpu_slot_preemption_timeout_ms]); global_context->setCPUSlotParking(new_server_settings[ServerSetting::cpu_slot_parking]); /// The thresholds were validated above, so a rejected reload never reaches here. Sets and stamps /// the shared thresholds, so every monitor adopts them on its next sample, /// bypassing the sticky cooldown. setMemoryPressureThresholds( new_server_settings[ServerSetting::reader_executor_memory_pressure_elevated_level_pct], new_server_settings[ServerSetting::reader_executor_memory_pressure_high_level_pct], new_server_settings[ServerSetting::reader_executor_memory_pressure_critical_level_pct]); if (config().has("resources") || config().has("workload_classifiers")) { LOG_WARNING( &logger(), "Config-based resource scheduling ('resources' and 'workload_classifiers' configuration sections) " "has been removed and is ignored. Use 'CREATE RESOURCE' and 'CREATE WORKLOAD' queries instead."); } /// Load WORKLOADs and RESOURCEs. global_context->getWorkloadEntityStoragePtr()->loadEntities(config()); if (!initial_loading) { /// We do not load ZooKeeper configuration on the first config loading /// because TestKeeper server is not started yet. if (zkutil::hasZooKeeperConfig(*reload_config)) global_context->reloadZooKeeperIfChanged(reload_config); global_context->reloadAuxiliaryZooKeepersConfigIfChanged(reload_config); global_context->reloadQueryMaskingRulesIfChanged(reload_config); if (global_context->isServerCompletelyStarted()) { std::lock_guard lock(servers_lock); updateServers(config(), new_server_settings, server_pool, *async_metrics, servers, servers_to_start_before_tables); } } global_context->updateStorageConfiguration(config()); global_context->updateInterserverCredentials(config()); { const size_t max_cache_size_in_bytes = static_cast( static_cast(current_physical_server_memory) * new_server_settings[ServerSetting::cache_size_to_ram_max_ratio]); global_context->updateUncompressedCacheConfiguration(config(), max_cache_size_in_bytes); global_context->updateColumnsCacheConfiguration( config(), getDefaultColumnsCacheSize(current_physical_server_memory, new_server_settings[ServerSetting::columns_cache_size_to_ram_ratio]), max_cache_size_in_bytes); if (auto columns_cache = global_context->getColumnsCache()) columns_cache->setAutoResizeSettings( new_server_settings[ServerSetting::columns_cache_free_memory_ratio], new_server_settings[ServerSetting::columns_cache_history_window_ms]); global_context->updateMarkCacheConfiguration(config(), max_cache_size_in_bytes); global_context->updateUniqueKeyIndexCacheConfiguration(config(), max_cache_size_in_bytes); global_context->updateDeleteBitmapCacheConfiguration(config(), max_cache_size_in_bytes); global_context->updatePrimaryIndexCacheConfiguration(config(), max_cache_size_in_bytes); global_context->updateStatisticsCacheConfiguration(config(), max_cache_size_in_bytes); global_context->updateIndexUncompressedCacheConfiguration(config(), max_cache_size_in_bytes); global_context->updateIndexMarkCacheConfiguration(config(), max_cache_size_in_bytes); global_context->updateVectorSimilarityIndexCacheConfiguration(config(), max_cache_size_in_bytes); global_context->updateTextIndexTokensCacheConfiguration(config(), max_cache_size_in_bytes); global_context->updateTextIndexHeaderCacheConfiguration(config(), max_cache_size_in_bytes); global_context->updateTextIndexPostingsCacheConfiguration(config(), max_cache_size_in_bytes); global_context->updateMMappedFileCacheConfiguration(config(), max_cache_size_in_bytes); global_context->updateQueryResultCacheConfiguration(config(), max_cache_size_in_bytes); global_context->updateQueryConditionCacheConfiguration(config(), max_cache_size_in_bytes); global_context->updateEncryptionHeaderCacheConfiguration(config(), max_cache_size_in_bytes); setPointInPolygonCacheMaxSizeInBytes( std::min(new_server_settings[ServerSetting::point_in_polygon_cache_size], max_cache_size_in_bytes)); #if USE_AVRO global_context->updateIcebergMetadataFilesCacheConfiguration(config(), max_cache_size_in_bytes); global_context->updatePaimonMetadataFilesCacheConfiguration(config(), max_cache_size_in_bytes); #endif #if USE_PARQUET global_context->updateParquetMetadataCacheConfiguration(config(), max_cache_size_in_bytes); #endif } #if USE_SSL CertificateReloader::instance().tryReloadAll(config()); #endif CompressionCodecEncrypted::Configuration::instance().tryLoad(config(), "encryption_codecs"); global_context->reloadStatelessWorkersConfig(config()); NamedCollectionFactory::instance().reloadFromConfig(config()); FileCacheFactory::instance().updateSettingsFromConfig(config()); HTTPConnectionPools::instance().setLimits( HTTPConnectionPools::Limits{ new_server_settings[ServerSetting::disk_connections_soft_limit], new_server_settings[ServerSetting::disk_connections_warn_limit], new_server_settings[ServerSetting::disk_connections_store_limit], new_server_settings[ServerSetting::disk_connections_hard_limit], }, HTTPConnectionPools::Limits{ new_server_settings[ServerSetting::storage_connections_soft_limit], new_server_settings[ServerSetting::storage_connections_warn_limit], new_server_settings[ServerSetting::storage_connections_store_limit], new_server_settings[ServerSetting::storage_connections_hard_limit], }, HTTPConnectionPools::Limits{ new_server_settings[ServerSetting::http_connections_soft_limit], new_server_settings[ServerSetting::http_connections_warn_limit], new_server_settings[ServerSetting::http_connections_store_limit], new_server_settings[ServerSetting::http_connections_hard_limit], }); HTTPConnectionPools::instance().setSocketBufferSizes( HTTPConnectionPools::SocketBufferSizes{ new_server_settings[ServerSetting::disk_connections_rcvbuf], new_server_settings[ServerSetting::disk_connections_sndbuf], }, HTTPConnectionPools::SocketBufferSizes{ new_server_settings[ServerSetting::storage_connections_rcvbuf], new_server_settings[ServerSetting::storage_connections_sndbuf], }, HTTPConnectionPools::SocketBufferSizes{ new_server_settings[ServerSetting::http_connections_rcvbuf], new_server_settings[ServerSetting::http_connections_sndbuf], }); DNSResolver::instance().setFilterSettings(new_server_settings[ServerSetting::dns_allow_resolve_names_to_ipv4], new_server_settings[ServerSetting::dns_allow_resolve_names_to_ipv6]); if (global_context->isServerCompletelyStarted()) CannotAllocateThreadFaultInjector::setFaultProbability(new_server_settings[ServerSetting::cannot_allocate_thread_fault_injection_probability]); /// Update core dump size limit. { rlimit rlim{}; if (getrlimit(RLIMIT_CORE, &rlim) == 0) { rlim.rlim_cur = config().getUInt64("core_dump.size_limit", 1024 * 1024 * 1024); if (setrlimit(RLIMIT_CORE, &rlim)) LOG_WARNING(log, "Cannot set max size of core file to {}", rlim.rlim_cur); } } ProfileEvents::increment(ProfileEvents::MainConfigLoads); /// Must be the last. latest_config = reload_config; }); const auto listen_hosts = getListenHosts(config()); const auto interserver_listen_hosts = getInterserverListenHosts(config()); const auto listen_try = getListenTry(config(), server_settings); if (config().has("keeper_server.server_id")) { #if USE_NURAFT //// If we don't have configured connection probably someone trying to use clickhouse-server instead //// of clickhouse-keeper, so start synchronously. bool can_initialize_keeper_async = false; if (has_zookeeper) /// We have configured connection to some zookeeper cluster { /// If we cannot connect to some other node from our cluster then we have to wait our Keeper start /// synchronously. can_initialize_keeper_async = global_context->tryCheckClientConnectionToMyKeeperCluster(); } /// Initialize certificates before Keeper RAFT so that NuRaft SSL contexts /// can register with CertificateReloader for hot-reload support. #if USE_SSL CertificateReloader::instance().tryLoad(config()); CertificateReloader::instance().tryLoadClient(config()); #endif /// Initialize keeper RAFT. global_context->initializeKeeperDispatcher(can_initialize_keeper_async); FourLetterCommandFactory::registerCommands(*global_context->getKeeperDispatcher()); auto config_getter = [this] () -> const Poco::Util::AbstractConfiguration & { return global_context->getConfigRef(); }; for (const auto & listen_host : listen_hosts) { /// TCP Keeper const char * port_name = "keeper_server.tcp_port"; createServer( config(), listen_host, port_name, listen_try, /* start_server: */ false, servers_to_start_before_tables, [&](UInt16 port) -> ProtocolServerAdapter { Poco::Net::ServerSocket socket; auto address = socketBindListen(server_settings, socket, listen_host, port); socket.setReceiveTimeout(Poco::Timespan(server_settings[ServerSetting::keeper_server_socket_receive_timeout_sec], 0)); socket.setSendTimeout(Poco::Timespan(server_settings[ServerSetting::keeper_server_socket_send_timeout_sec], 0)); return ProtocolServerAdapter( listen_host, port_name, ServerType::Type::END, "Keeper (tcp): " + address.toString(), std::make_unique( new KeeperTCPHandlerFactory( config_getter, global_context->getKeeperDispatcher(), global_context->getSettingsRef()[Setting::receive_timeout].totalSeconds(), global_context->getSettingsRef()[Setting::send_timeout].totalSeconds(), false), server_pool, socket)); }); const char * secure_port_name = "keeper_server.tcp_port_secure"; createServer( config(), listen_host, secure_port_name, listen_try, /* start_server: */ false, servers_to_start_before_tables, [&](UInt16 port) -> ProtocolServerAdapter { #if USE_SSL Poco::Net::SecureServerSocket socket; auto address = socketBindListen(server_settings, socket, listen_host, port, /* secure = */ true); socket.setReceiveTimeout(Poco::Timespan(server_settings[ServerSetting::keeper_server_socket_receive_timeout_sec], 0)); socket.setSendTimeout(Poco::Timespan(server_settings[ServerSetting::keeper_server_socket_send_timeout_sec], 0)); return ProtocolServerAdapter( listen_host, secure_port_name, ServerType::Type::END, "Keeper with secure protocol (tcp_secure): " + address.toString(), std::make_unique( new KeeperTCPHandlerFactory( config_getter, global_context->getKeeperDispatcher(), global_context->getSettingsRef()[Setting::receive_timeout].totalSeconds(), global_context->getSettingsRef()[Setting::send_timeout].totalSeconds(), true), server_pool, socket)); #else UNUSED(port); throw Exception(ErrorCodes::SUPPORT_IS_DISABLED, "SSL support for TCP protocol is disabled because Poco library was built without NetSSL support."); #endif }); /// HTTP control endpoints port_name = "keeper_server.http_control.port"; if (!config().getString(port_name, "").empty()) { auto handler_factory = createKeeperHTTPHandlerFactory( *this, config_getter(), global_context->getKeeperDispatcher(), "KeeperHTTPHandler-factory"); createServer(config(), listen_host, port_name, listen_try, /* start_server: */ false, servers_to_start_before_tables, [&](UInt16 port) -> ProtocolServerAdapter { auto http_context = httpContext(); Poco::Timespan keep_alive_timeout(server_settings[ServerSetting::keep_alive_timeout].totalSeconds(), 0); Poco::Net::HTTPServerParams::Ptr http_params = new Poco::Net::HTTPServerParams; http_params->setTimeout(http_context->getReceiveTimeout()); http_params->setKeepAliveTimeout(keep_alive_timeout); Poco::Net::ServerSocket socket; auto address = socketBindListen(server_settings, socket, listen_host, port); socket.setReceiveTimeout(http_context->getReceiveTimeout()); socket.setSendTimeout(http_context->getSendTimeout()); return ProtocolServerAdapter( listen_host, port_name, ServerType::Type::END, "HTTP Control: http://" + address.toString(), std::make_unique( std::move(http_context), handler_factory, server_pool, socket, http_params)); }); } /// HTTPS control endpoints port_name = "keeper_server.http_control.secure_port"; if (!config().getString(port_name, "").empty()) { #if USE_SSL auto handler_factory = createKeeperHTTPHandlerFactory( *this, config_getter(), global_context->getKeeperDispatcher(), "KeeperHTTPSHandler-factory"); #endif createServer(config(), listen_host, port_name, listen_try, /* start_server: */ false, servers_to_start_before_tables, [&](UInt16 port) -> ProtocolServerAdapter { #if USE_SSL auto http_context = httpContext(); Poco::Timespan keep_alive_timeout(server_settings[ServerSetting::keep_alive_timeout].totalSeconds(), 0); Poco::Net::HTTPServerParams::Ptr http_params = new Poco::Net::HTTPServerParams; http_params->setTimeout(http_context->getReceiveTimeout()); http_params->setKeepAliveTimeout(keep_alive_timeout); Poco::Net::SecureServerSocket socket; auto address = socketBindListen(server_settings, socket, listen_host, port, /* secure = */ true); socket.setReceiveTimeout(http_context->getReceiveTimeout()); socket.setSendTimeout(http_context->getSendTimeout()); return ProtocolServerAdapter( listen_host, port_name, ServerType::Type::END, "HTTPS Control: https://" + address.toString(), std::make_unique( std::move(http_context), handler_factory, server_pool, socket, http_params)); #else UNUSED(port); throw Exception(ErrorCodes::SUPPORT_IS_DISABLED, "HTTPS control protocol is disabled because Poco library was built without NetSSL support."); #endif }); } } #else throw Exception(ErrorCodes::SUPPORT_IS_DISABLED, "ClickHouse server built without NuRaft library. Cannot use internal coordination."); #endif } { std::lock_guard lock(servers_lock); /// We should start interserver communications before (and more important shutdown after) tables. /// Because server can wait for a long-running queries (for example in tcp_handler) after interserver handler was already shut down. /// In this case we will have replicated tables which are unable to send any parts to other replicas, but still can /// communicate with zookeeper, execute merges, etc. createInterserverServers( config(), server_settings, interserver_listen_hosts, listen_try, server_pool, *async_metrics, servers_to_start_before_tables, /* start_servers= */ false); } #if USE_SSL /// We may notice that try to reload certificates twice within this function. /// First time here before starting the `servers_to_start_before_tables`, and then /// one more time before starting the `servers`. The problem we're trying to avoid is: /// 1. We create a socket for keeper secure protocol (w/o starting a server yet). That creates a SSL context (in particular, it might be the default context). /// 2. We then start this server a few lines below. While handling the first request, it will access (read) certs. /// 3. A little further down, we reload certificates before starting the rest of the servers. /// 4. `CertificateReloader` will set its custom `cert_cb` to the default context if it is not initialized yet (see `CertificateReloader::init()`). /// 5. Items (2) and (4) are not synchronized, so there might be a data race for example (see https://github.com/ClickHouse/ClickHouse/issues/85412). CertificateReloader::instance().tryLoad(config()); CertificateReloader::instance().tryLoadClient(config()); #endif { std::lock_guard lock(servers_lock); startServers(servers_to_start_before_tables, listen_try, log); for (const auto & server : servers_to_start_before_tables) { if (server.bindsOnStart()) global_context->registerServerPort(server.getPortName(), static_cast(config().getInt(server.getPortName()))); } } #if USE_SSL /// ACME client is necessary for CertificateReloader to work in case Let's Encrypt is configured, /// but can not start before Keeper server can be initialized (in case of embedded Keeper). /// Let's try deferring until servers are started. ACME::Client::instance().initialize(config()); #endif const auto & cache_disk_name = server_settings[ServerSetting::temporary_data_in_cache].value; /// Storage with temporary data for processing of heavy queries. if (!server_settings[ServerSetting::tmp_policy].value.empty()) { global_context->setTemporaryStoragePolicy( server_settings[ServerSetting::tmp_policy], server_settings[ServerSetting::max_temporary_data_on_disk_size]); } else if (!cache_disk_name.empty()) { /// When cache is used as a temporary data storage with load_metadata_asynchronously /// the cache might not be fully ready when setTemporaryStorageInCache() is called /// below which might try to use the cache before it's fully loaded leading to /// `File cache is not initialized` errors. Instead disallow load_metadata_asynchronously /// for cache disks used as temporary storage. auto cache_name = global_context->getDisk(cache_disk_name)->getCacheName(); if (!cache_name.empty()) { auto cache_data = FileCacheFactory::instance().getByName(cache_name); if (cache_data->getSettings()[FileCacheSetting::load_metadata_asynchronously]) { throw Exception( ErrorCodes::INVALID_SETTING_VALUE, "Cache disk '{}' is used as a temporary storage, load_metadata_asynchronously is disallowed.", cache_name); } } global_context->setTemporaryStorageInCache( server_settings[ServerSetting::temporary_data_in_cache], server_settings[ServerSetting::max_temporary_data_on_disk_size]); } else { const auto & tmp_path_setting = server_settings[ServerSetting::tmp_path]; std::string temporary_path = tmp_path_setting.changed ? getCanonicalPath(String(tmp_path_setting.value), path_str) : String(path / "tmp/"); global_context->setTemporaryStoragePath(temporary_path, server_settings[ServerSetting::max_temporary_data_on_disk_size]); } /// Initialize access storages. auto & access_control = global_context->getAccessControl(); try { access_control.setupFromMainConfig(config(), config_path, [&] { return global_context->getZooKeeper(); }); } catch (...) { tryLogCurrentException(log, "Caught exception while setting up access control."); throw; } auto disable_config_reload_callbacks = [&] { global_context->setConfigReloadCallback([] { throw Exception(ErrorCodes::ABORTED, "Cannot reload config because the server is shutting down"); }); if (cgroups_memory_usage_observer) cgroups_memory_usage_observer->setOnMemoryAmountAvailableChangedFn([] {}); }; SCOPE_EXIT({ disable_config_reload_callbacks(); }); if (cgroups_memory_usage_observer) { cgroups_memory_usage_observer->setOnMemoryAmountAvailableChangedFn([&]() { main_config_reloader->reload(); }); cgroups_memory_usage_observer->startThread(); } /// Reload config in SYSTEM RELOAD CONFIG query. global_context->setConfigReloadCallback([&]() { main_config_reloader->reload(); access_control.reload(AccessControl::ReloadMode::USERS_CONFIG_ONLY); }); global_context->setStopServersCallback([&](const ServerType & server_type) { std::lock_guard lock(servers_lock); stopServers(servers, server_type); }); global_context->setStartServersCallback([&](const ServerType & server_type) { std::lock_guard lock(servers_lock); /// Introspection server can run SYSTEM START LISTEN before other servers' startup or after other servers' shutdown. /// In either case, we need to return an error. if (is_cancelled || !global_context->isServerCompletelyStarted()) throw Exception(ErrorCodes::ABORTED, "Cannot start listeners because the server is starting up or shutting down"); /// The config may have been reloaded since startup, so recompute the listener /// configuration from the current config instead of the startup snapshot. ServerSettings current_server_settings; current_server_settings.loadSettingsFromConfig(config()); createServers( config(), current_server_settings, getListenHosts(config()), getListenTry(config(), current_server_settings), server_pool, *async_metrics, servers, /* start_servers= */ true, server_type); }); /// Limit on total number of concurrently executed queries. global_context->getProcessList().setMaxSize(server_settings[ServerSetting::max_concurrent_queries]); /// Load global settings from default_profile and system_profile. global_context->setDefaultProfiles(config()); /// Initialize background executors after we load default_profile config. /// This is needed to load proper values of background_pool_size etc. global_context->initializeBackgroundExecutorsIfNeeded(); if (server_settings[ServerSetting::async_insert_threads]) { global_context->setAsynchronousInsertQueue(std::make_shared( global_context, server_settings[ServerSetting::async_insert_threads], server_settings[ServerSetting::async_insert_queue_flush_on_shutdown])); } /// Set path for format schema files const auto & format_schema_path_setting = server_settings[ServerSetting::format_schema_path]; fs::path format_schema_path(format_schema_path_setting.changed ? fs::path(getCanonicalPath(String(format_schema_path_setting.value), path_str)) : path / "format_schemas/"); global_context->setFormatSchemaPath(format_schema_path); fs::create_directories(format_schema_path); /// Set the path for google proto files if (server_settings[ServerSetting::google_protos_path].changed) global_context->setGoogleProtosPath(fs::weakly_canonical(server_settings[ServerSetting::google_protos_path].value)); /// Set path for filesystem caches { String filesystem_caches_path = server_settings[ServerSetting::filesystem_caches_path].value; if (!filesystem_caches_path.empty()) global_context->setFilesystemCachesPath(getCanonicalPath(std::move(filesystem_caches_path), path_str)); } /// NOTE: Do sanity checks after we loaded all possible substitutions (for the configuration) from ZK /// Additionally, making the check after the default profile is initialized. /// It is important to initialize MergeTreeSettings after Settings, to support compatibility for MergeTreeSettings. addEnvironmentWarnings(global_context, logger(), path_str, server_settings[ServerSetting::logger_log]); /// Check sanity of MergeTreeSettings on server startup { size_t background_pool_tasks = global_context->getMergeMutateExecutor()->getMaxTasksCount(); global_context->getMergeTreeSettings().sanityCheck(background_pool_tasks, global_context->wasBackgroundPoolAutoLowered()); global_context->getReplicatedMergeTreeSettings().sanityCheck( background_pool_tasks, global_context->wasBackgroundPoolAutoLowered()); } /// try set up encryption. There are some errors in config, error will be printed and server wouldn't start. CompressionCodecEncrypted::Configuration::instance().load(config(), "encryption_codecs"); auto replicas_reconnector = ReplicasReconnector::init(global_context); /// Set current database name before loading tables and databases because /// system logs may copy global context. std::string default_database = server_settings[ServerSetting::default_database]; if (default_database.empty()) throw Exception(ErrorCodes::BAD_ARGUMENTS, "default_database cannot be empty"); global_context->setCurrentDatabaseNameInGlobalContext(default_database); /// Start collecting asynchronous metrics before loading tables, so that the Prometheus endpoint /// (started below) exposes meaningful values during the potentially long metadata loading phase. /// This includes the OS/jemalloc metrics and the per-table metrics such as /// `TotalIndexGranularityBytesInMemoryAllocated`, which let one observe memory growth as tables /// are loaded. The metric computation skips not-yet-loaded tables (just like with /// `async_load_databases`), and writing to `system.asynchronous_metric_log` is skipped until the /// system logs are initialized below. The asynchronous metrics thread reads the `servers` lists /// under `servers_lock`, so it is safe to start before the main `servers` exist. async_metrics->start(); global_context->setAsynchronousMetrics(async_metrics.get()); /// Start the Prometheus endpoint before loading tables, so that metrics stay observable during /// the potentially long metadata loading phase. Only do it for metrics-only configurations: /// custom `prometheus.handlers` may serve queries (`remote_write`, `remote_read`, `query`, /// `api_v1`) and must follow the regular `servers` lifecycle. The server is created in the /// regular `servers` list, so runtime reconfiguration and `SYSTEM START/STOP LISTEN` handle it /// as usual. The later `createServers` call skips it (`createServer` does not recreate a live /// server), and the start loop for `servers` skips it too (`ProtocolServerAdapter::start` does /// nothing for an already started server). if (!config().has("prometheus.handlers")) { std::lock_guard lock(servers_lock); createServers( config(), server_settings, listen_hosts, listen_try, server_pool, *async_metrics, servers, /* start_servers= */ true, ServerType(ServerType::Type::PROMETHEUS)); } LOG_INFO(log, "Loading metadata from {}", path_str); LoadTaskPtrs load_system_metadata_tasks; LoadTaskPtrs load_metadata_tasks; // Make sure that if exception is thrown during startup async, new async loading jobs are not going to be called. // This is important for the case when exception is thrown from loading of metadata with `async_load_databases = false` // to avoid simultaneously running table startups and destructing databases. SCOPE_EXIT_SAFE( LOG_INFO(log, "Stopping AsyncLoader."); // Waits for all currently running jobs to finish and do not run any other pending jobs. global_context->getAsyncLoader().shutdown(); ); /// The introspection servers accept TCP connections while the server is otherwise unreachable: /// before attaching and after detaching of tables. if (hasIntrospectionProtocols(config())) { global_context->setStopIntrospectionServersCallback([&] { if (introspection_servers.empty()) return; LOG_DEBUG(log, "Waiting for current connections to introspection servers to finish."); size_t current_connections = 0; { std::lock_guard lock(servers_lock); for (auto & server : introspection_servers) { server.stop(); current_connections += server.currentConnections(); } } if (current_connections) LOG_INFO(log, "Closed all introspection listening sockets. Waiting for {} outstanding connections.", current_connections); else LOG_INFO(log, "Closed all introspection listening sockets."); global_context->getProcessList().killAllQueries(); if (current_connections) current_connections = waitServersToFinish( introspection_servers, servers_lock, server_settings[ServerSetting::shutdown_wait_unfinished]); if (current_connections) { dumpCoverageReportIfPossible(); LOG_WARNING(log, "Closed connections to introspection servers. But {} remain. Will shutdown forcefully.", current_connections); /// No leak check: remaining connection handlers still own memory it would report as /// leaked. Reported, because here stderr is the server log. safeExit(0, LeakCheck::SkipAndReport); } LOG_INFO(log, "Closed connections to introspection servers."); introspection_server_pool->joinAll(); }); introspection_server_pool = std::make_unique( /* minCapacity */ 1, /* maxCapacity */ server_settings[ServerSetting::max_connections], /* idleTime */ 60, /* stackSize */ DEFAULT_THREAD_STACK_SIZE ? static_cast(DEFAULT_THREAD_STACK_SIZE) : POCO_THREAD_STACK_SIZE, server_settings[ServerSetting::global_profiler_real_time_period_ns], server_settings[ServerSetting::global_profiler_cpu_time_period_ns]); std::lock_guard lock(servers_lock); createServers( config(), server_settings, listen_hosts, listen_try, *introspection_server_pool, *async_metrics, introspection_servers, /* start_servers= */ true, ServerType(ServerType::Type::QUERIES_ALL), /* only_introspection_protocols= */ true); } try { /// Don't run background queries until we loaded tables. /// (In particular things would break if a background drop query happens before the /// loadMarkedAsDroppedTables() call below - it'll see dropped table metadata and try to /// drop the table a second time and throw an exception.) global_context->getRefreshSet().setRefreshesStopped(true); auto & database_catalog = DatabaseCatalog::instance(); /// We load temporary database first, because projections need it. database_catalog.initializeAndLoadTemporaryDatabase(); /// The tables (including the ones in the `system` database) can depend on the dictionaries from the configuration files, /// so register them before loading the databases. They start loading later, in `loadOrReloadDictionaries`. global_context->registerDictionariesBeforeLoadingMetadata(config()); load_system_metadata_tasks = loadMetadataSystem(global_context, server_settings[ServerSetting::async_load_system_database]); maybeConvertSystemDatabase(global_context, load_system_metadata_tasks); /// Startup scripts can depend on the system log tables. if (config().has("startup_scripts") && !server_settings[ServerSetting::prepare_system_log_tables_on_startup].changed) global_context->setServerSetting("prepare_system_log_tables_on_startup", true); /// After attaching system databases we can initialize system log. global_context->initializeSystemLogs(); if (has_trace_collector) global_context->initializeTraceCollector(); #if defined(OS_LINUX) if (oom_canary) oom_canary->start(); #endif global_context->handleSystemZooKeeperConnectionLogAfterInitializationIfNeeded(); /// Build loggers before tables startup to make log messages from tables /// attach available in system.text_log buildLoggers(config(), logger()); initializeAzureSDKLogger(server_settings, logger().getLevel()); /// After the system database is created, attach virtual system tables (in addition to query_log and part_log) bool has_keeper_server = false; #if USE_NURAFT has_keeper_server = global_context->tryGetKeeperDispatcher() != nullptr; #endif attachSystemTablesServer(global_context, *database_catalog.getSystemDatabase(), has_zookeeper, has_keeper_server); attachInformationSchema(global_context, *database_catalog.getDatabase(DatabaseCatalog::INFORMATION_SCHEMA)); attachInformationSchema(global_context, *database_catalog.getDatabase(DatabaseCatalog::INFORMATION_SCHEMA_UPPERCASE)); /// Firstly remove partially dropped databases, to avoid race with Materialized...SyncThread, /// that may execute DROP before loadMarkedAsDroppedTables() in background, /// and so loadMarkedAsDroppedTables() will find it and try to add, and UUID will overlap. database_catalog.loadMarkedAsDroppedTables(); database_catalog.createBackgroundTasks(); /// Then, load remaining databases (some of them maybe be loaded asynchronously) load_metadata_tasks = loadMetadata(global_context, default_database, server_settings[ServerSetting::async_load_databases]); /// If we need to convert database engines, disable async tables loading convertDatabasesEnginesIfNeed(load_metadata_tasks, global_context); database_catalog.startupBackgroundTasks(); /// After loading validate that default database exists database_catalog.assertDatabaseExists(default_database); /// Load user-defined SQL functions. global_context->loadUserDefinedExecutableFunctionDrivers(config()); global_context->getUserDefinedSQLObjectsStorage().loadObjects(); /// For driver-based executable UDFs persisted as ATTACH FUNCTION queries, ensure the /// dynamic configuration files exist; re-run their drivers if they are missing. UserDefinedSQLFunctionFactory::instance().reloadDriverBasedFunctions( global_context, global_context->getUserDefinedSQLObjectsStorage()); global_context->getRefreshSet().setRefreshesStopped(false); } catch (...) { tryLogCurrentException(log, "Caught exception while loading metadata"); throw; } DatabaseCatalog::instance().startReplicatedDDLQueries(); LOG_DEBUG(log, "Loaded metadata."); #if defined(OS_LINUX) auto tasks_stats_provider = TasksStatsCounters::findBestAvailableProvider(); if (tasks_stats_provider == TasksStatsCounters::MetricsProvider::None) { LOG_INFO(log, "It looks like this system does not have procfs mounted at /proc location." " 'taskstats' performance statistics will be disabled." " It could happen due to incorrect ClickHouse package installation."); } else { LOG_INFO(log, "Tasks stats provider: {}", TasksStatsCounters::metricsProviderString(tasks_stats_provider)); } if (!hasLinuxCapability(CAP_SYS_NICE)) { LOG_INFO(log, "It looks like the process has no CAP_SYS_NICE capability, the setting 'os_thread_priority' will have no effect." " It could happen due to incorrect ClickHouse package installation." " You could resolve the problem manually with 'sudo setcap cap_sys_nice=+ep {}'." " Note that it will not work on 'nosuid' mounted filesystems.", executable_path); } #else LOG_INFO(log, "TaskStats is not implemented for this OS. IO accounting will be disabled."); #endif { attachSystemTablesAsync(global_context, *DatabaseCatalog::instance().getSystemDatabase(), *async_metrics); { std::lock_guard lock(servers_lock); createServers(config(), server_settings, listen_hosts, listen_try, server_pool, *async_metrics, servers); if (servers.empty()) throw Exception(ErrorCodes::NO_ELEMENTS_IN_CONFIG, "No servers started (add valid listen_host and 'tcp_port' or 'http_port' " "to configuration file.)"); } if (servers.empty()) throw Exception(ErrorCodes::NO_ELEMENTS_IN_CONFIG, "No servers started (add valid listen_host and 'tcp_port' or 'http_port' " "to configuration file.)"); #if USE_SSL ACME::Client::instance().initialize(config()); CertificateReloader::instance().tryLoad(config()); CertificateReloader::instance().tryLoadClient(config()); #endif main_config_reloader->start(); access_control.startPeriodicReloading(); /// try to load dictionaries immediately, throw on error and die try { global_context->loadOrReloadDictionaries(config()); if (!server_settings[ServerSetting::dictionaries_lazy_load] && server_settings[ServerSetting::wait_dictionaries_load_at_startup]) global_context->waitForDictionariesLoad(); } catch (...) { tryLogCurrentException(log, "Caught exception while loading dictionaries."); throw; } /// try to load embedded dictionaries immediately, throw on error and die try { global_context->tryCreateEmbeddedDictionaries(); } catch (...) { tryLogCurrentException(log, "Caught exception while loading embedded dictionaries."); throw; } /// try to load user defined executable functions, throw on error and die try { global_context->loadOrReloadUserDefinedExecutableFunctions(config()); } catch (...) { tryLogCurrentException(log, "Caught exception while loading user defined executable functions."); throw; } if (has_zookeeper && config().has("distributed_ddl")) { /// DDL worker should be started after all tables were loaded String ddl_queue_path = server_settings[ServerSetting::distributed_ddl_path]; String ddl_replicas_path = server_settings[ServerSetting::distributed_ddl_replicas_path]; int pool_size = server_settings[ServerSetting::distributed_ddl_pool_size]; if (pool_size < 1) throw Exception(ErrorCodes::ARGUMENT_OUT_OF_BOUND, "distributed_ddl.pool_size should be greater then 0"); global_context->setDDLWorker( std::make_unique( pool_size, ddl_queue_path, ddl_replicas_path, global_context, &config(), "distributed_ddl", "default", "DDLWorker", &CurrentMetrics::MaxDDLEntryID, &CurrentMetrics::MaxPushedDDLEntryID), joinTasks(load_system_metadata_tasks, load_metadata_tasks)); } /// Do not keep tasks in server, they should be kept inside databases. Used here to make dependent tasks only. load_system_metadata_tasks.clear(); load_system_metadata_tasks.shrink_to_fit(); load_metadata_tasks.clear(); load_metadata_tasks.shrink_to_fit(); if (config().has("startup_scripts")) loadStartupScripts(config(), server_settings, global_context, log); auto stop_acme_instance = []{ #if USE_SSL /// Stop ACME tasks. ACME::Client::instance().shutdown(); #endif }; /// Wrapping the call to OOM canary stop in a lambda lets us write /// the OS_LINUX guard outside the SCOPE_EXIT_SAFE macro argument list, /// avoiding -Wembedded-directive. auto stop_oom_canary = [&]{ #if defined(OS_LINUX) if (oom_canary) oom_canary->stop(); #endif }; SCOPE_EXIT_SAFE({ const auto & logger_shutdown_level_setting = server_settings[ServerSetting::logger_shutdown_level]; if (logger_shutdown_level_setting.changed && !logger_shutdown_level_setting.value.empty()) { /// Set the root logger level to the shutdown level. /// This is useful for debugging shutdown issues. config().setString("logger.level", logger_shutdown_level_setting.value); Loggers::updateLevels(config(), logger()); LOG_INFO(log, "Set root logger in level {} before shutdown", logger_shutdown_level_setting.value); } if (config().has("logger.shutdown_console_log_level") && !config().getString("logger.shutdown_console_log_level").empty()) { /// Set the root logger level to the shutdown level. /// This is useful for debugging shutdown issues. config().setString("logger.console_log_level", config().getString("logger.shutdown_console_log_level")); Loggers::updateLevels(config(), logger()); LOG_INFO(log, "Set console logger in level {} before shutdown", config().getString("logger.shutdown_console_log_level")); } LOG_DEBUG(log, "Received termination signal."); CurrentMetrics::set(CurrentMetrics::IsServerShuttingDown, 1); /// Stop reloading of the main config. This must be done before everything else because it /// can try to access/modify already deleted objects. /// E.g. it can recreate new servers or it may pass a changed config to some destroyed parts of ContextSharedPart. disable_config_reload_callbacks(); main_config_reloader.reset(); access_control.stopPeriodicReloading(); stop_acme_instance(); is_cancelled = true; LOG_DEBUG(log, "Waiting for current connections to close."); size_t current_connections = 0; { std::lock_guard lock(servers_lock); for (auto & server : servers) { server.stop(); current_connections += server.currentConnections(); } } global_context->getBackgroundQueryPool().finish(); global_context->getRefreshSet().setRefreshesStopped(true); if (current_connections) LOG_WARNING(log, "Closed all listening sockets. Waiting for {} outstanding connections.", current_connections); else LOG_INFO(log, "Closed all listening sockets."); /// Wait for unfinished backups and restores. /// This must be done after closing listening sockets (no more socket-delivered backups/restores) but before ProcessList::killAllQueries /// (because killAllQueries() will cancel all running backups/restores). bool backups_finished = true; if (server_settings[ServerSetting::shutdown_wait_backups_and_restores]) global_context->waitAllBackupsAndRestores(); else { /// Refused first so that the wait cannot miss an operation started after it took /// its snapshot; a distributed DDL query can still deliver one here. global_context->stopAcceptingNewBackupsAndRestores(); backups_finished = global_context->cancelAllBackupsAndRestores( std::chrono::steady_clock::now() + std::chrono::seconds(server_settings[ServerSetting::shutdown_wait_unfinished])); } stop_oom_canary(); /// Killing remaining queries. /// In wait mode the DDL worker's parts of ON CLUSTER backups and restores are spared, and waited for below. /// Any other backup or restore that missed the wait above is cancelled: an initiator's own part needs this host's DDL worker. const bool wait_backups_and_restores = server_settings[ServerSetting::shutdown_wait_backups_and_restores]; if (!server_settings[ServerSetting::shutdown_wait_unfinished_queries]) global_context->getProcessList().killAllQueries(/* except_ddl_backups_and_restores= */ wait_backups_and_restores); size_t wait_limit_seconds = server_settings[ServerSetting::shutdown_wait_unfinished]; auto wait_start = std::chrono::steady_clock::now(); auto shutdown_deadline = wait_start + std::chrono::milliseconds(wait_limit_seconds * 1000); if (current_connections) current_connections = waitServersToFinish(servers, servers_lock, wait_limit_seconds); if (current_connections) LOG_WARNING(log, "Closed connections. But {} remain." " Tip: To increase wait time add to config: 300", current_connections); else LOG_INFO(log, "Closed connections."); bool joined_refresh_tasks = global_context->getRefreshSet().joinBackgroundTasks(shutdown_deadline); bool joined_background_queries = global_context->getBackgroundQueryPool().waitUntil(shutdown_deadline); if (!joined_background_queries) LOG_WARNING(log, "{} background queries remain.", global_context->getBackgroundQueryPool().active()); dns_cache_updater.reset(); /// killAllQueries() and the waits above can have driven a cancelled backup to a final /// status, and then the normal teardown is able to complete. Sound because this is false /// only where new operations are already refused, so the unfinished set cannot grow. if (!backups_finished) backups_finished = !global_context->hasUnfinishedBackupsAndRestores(); if (current_connections || !joined_refresh_tasks || !joined_background_queries || !backups_finished) { /// There is no better way to force connections to close in Poco. /// Otherwise connection handlers will continue to live /// (they are effectively dangling objects, but they use global thread pool /// and global thread pool destructor will wait for threads, preventing server shutdown). /// Dump coverage here, because std::atexit callback would not be called. dumpCoverageReportIfPossible(); LOG_WARNING(log, "Will shutdown forcefully."); /// No leak check: remaining connection handlers, refresh tasks or background queries still own memory /// it would report as leaked. Reported, because here stderr is the server log. safeExit(0, LeakCheck::SkipAndReport); } /// Before the AsyncLoader stops, because a restore can create a database, and after the forced exit, because stopping /// the DDL worker waits for the task it runs, which the kill cancels unless it is such a part. if (wait_backups_and_restores) { global_context->getProcessList().killAllQueries(/* except_ddl_backups_and_restores= */ true); global_context->stopDDLWorkerAndWaitAllBackupsAndRestores(); } }); { std::lock_guard lock(servers_lock); /// Restore the startup log level overrides before accepting connections, /// so that no requests are served with an elevated (startup) log level. /// This must happen before server.start() because the config reload callback /// (ConfigReloader) reads from config() which includes the writable layer /// where startup level overrides are stored. if (should_restore_default_logger_level) { config().setString("logger.level", default_logger_level_config); Loggers::updateLevels(config(), logger()); LOG_INFO(log, "Restored default logger level to {}", default_logger_level_config); } if (should_restore_console_log_level) { /// If the level was unset just remove the override so the default can be set via /// Loggers::updateLevels again; otherwise restore the configured value. if (console_log_level_was_set) { config().setString("logger.console_log_level", original_console_log_level_config); Loggers::updateLevels(config(), logger()); LOG_INFO(log, "Restored console logger level to {}", original_console_log_level_config); } else { config().remove("logger.console_log_level"); Loggers::updateLevels(config(), logger()); LOG_INFO(log, "Restored console logger level to logger.level"); } } startServers(servers, listen_try, log); for (const auto & server : servers) { if (server.bindsOnStart()) global_context->registerServerPort(server.getPortName(), static_cast(config().getInt(server.getPortName()))); } if (servers.empty()) throw Exception(ErrorCodes::NO_ELEMENTS_IN_CONFIG, "No servers started (add valid listen_host and 'tcp_port' or 'http_port' to configuration file.)"); global_context->setServerCompletelyStarted(); LOG_INFO(log, "Ready for connections."); } startup_watch.stop(); ProfileEvents::increment(ProfileEvents::ServerStartupMilliseconds, startup_watch.elapsedMilliseconds()); CannotAllocateThreadFaultInjector::setFaultProbability(server_settings[ServerSetting::cannot_allocate_thread_fault_injection_probability]); try { global_context->startClusterDiscovery(); } catch (...) { tryLogCurrentException(log, "Caught exception while starting cluster discovery"); } #if defined(OS_LINUX) /// Tell the service manager that service startup is finished. /// NOTE: the parent clickhouse-watchdog process must do systemdNotify("MAINPID={}\n", child_pid); before /// the child process notifies 'READY=1'. systemdNotify("READY=1\n"); #endif std::vector> metrics_transmitters; for (const auto & graphite_key : DB::getMultipleKeysFromConfig(config(), "", "graphite")) { metrics_transmitters.emplace_back(std::make_unique( global_context->getConfigRef(), graphite_key, *async_metrics)); } waitForTerminationRequest(); } return Application::EXIT_OK; } catch (...) { /// Poco does not provide stacktrace. tryLogCurrentException("Application"); auto code = getCurrentExceptionCode(); return static_cast(code) ? code : -1; } namespace { /// Maps a `` stack onto the ServerType of its innermost layer, so that a composable /// protocol is reported under the same metrics as the equivalent built-in port. ServerType::Type protocolStackType(const std::string & leaf_type, bool is_secure, bool has_proxy) { if (leaf_type == "tcp") { if (is_secure) return ServerType::Type::TCP_SECURE; return has_proxy ? ServerType::Type::TCP_WITH_PROXY : ServerType::Type::TCP; } if (leaf_type == "http") return is_secure ? ServerType::Type::HTTPS : ServerType::Type::HTTP; if (leaf_type == "interserver") return is_secure ? ServerType::Type::INTERSERVER_HTTPS : ServerType::Type::INTERSERVER_HTTP; if (leaf_type == "mysql") return ServerType::Type::MYSQL; if (leaf_type == "postgres") return ServerType::Type::POSTGRESQL; if (leaf_type == "prometheus") return ServerType::Type::PROMETHEUS; return ServerType::Type::END; } /// Walk a composable protocol's `impl` chain and return the effective `default_session_user`: /// the value closest to the endpoint wins. Used both when the protocol stack is built and when it is /// decided whether a configuration reload has to restart the endpoint. /// Returns an empty optional if no module in the chain sets it (the handlers then fall back to the /// `default_session_user` server setting). std::optional getEffectiveDefaultSessionUser(const Poco::Util::AbstractConfiguration & config, const std::string & protocol) { std::string conf_name = protocol; std::string prefix = protocol + "."; std::unordered_set pset {conf_name}; while (true) { if (config.has(prefix + "default_session_user")) return config.getString(prefix + "default_session_user"); if (!config.has(prefix + "impl")) return {}; conf_name = "protocols." + config.getString(prefix + "impl"); prefix = conf_name + "."; /// A malformed loop is rejected when the stack is built; here we just stop to avoid spinning. if (!pset.insert(conf_name).second) return {}; } } /// Resolve the `http_handlers`-style section a composable `http` endpoint serves: the /// `handlers` key of the module carrying `type = http`, found by walking the `impl` chain /// (mirrors `buildProtocolStackFromConfig`). Returns an empty optional when the chain does /// not reach an `http` module in this configuration (e.g. the endpoint did not exist or had /// a different type before a reload). std::optional resolveHTTPHandlersKey(const Poco::Util::AbstractConfiguration & config, const std::string & protocol) { std::string conf_name = protocol; std::string prefix = protocol + "."; std::unordered_set pset {conf_name}; while (true) { if (config.getString(prefix + "type", "") == "http") return config.getString(conf_name + ".handlers", "http_handlers"); if (!config.has(prefix + "impl")) return {}; conf_name = "protocols." + config.getString(prefix + "impl"); prefix = conf_name + "."; /// A malformed loop is rejected when the stack is built; here we just stop to avoid spinning. if (!pset.insert(conf_name).second) return {}; } } /// The ``-style sections the HTTP listeners of a configuration serve: the default /// `http_handlers` for `http_port` and `https_port`, and the section each composable `http` endpoint /// references. A section no listener serves is left out, so that validating a configuration accepts /// exactly what starting the server with it would. Strings servedHTTPHandlersKeys(const Poco::Util::AbstractConfiguration & config) { std::unordered_set keys; if (config.has("http_port") || config.has("https_port")) keys.insert("http_handlers"); if (config.has("protocols")) { Poco::Util::AbstractConfiguration::Keys protocols; config.keys("protocols", protocols); for (const auto & protocol : protocols) { if (auto handlers_key = resolveHTTPHandlersKey(config, "protocols." + protocol)) keys.insert(*handlers_key); } } return {keys.begin(), keys.end()}; } /// Whether a listener of this configuration serves the `prometheus` section on a port of its own: the /// standalone `prometheus.port` listener, or a composable `type = prometheus` endpoint (whose type is /// found by walking the `impl` chain, as in `buildProtocolStackFromConfig`). bool hasPrometheusListener(const Poco::Util::AbstractConfiguration & config) { if (config.getInt("prometheus.port", 0)) return true; if (!config.has("protocols")) return false; Poco::Util::AbstractConfiguration::Keys protocols; config.keys("protocols", protocols); for (const auto & protocol : protocols) { std::string conf_name = "protocols." + protocol; std::string prefix = conf_name + "."; std::unordered_set pset {conf_name}; while (true) { if (config.getString(prefix + "type", "") == "prometheus") return true; if (!config.has(prefix + "impl")) break; conf_name = "protocols." + config.getString(prefix + "impl"); prefix = conf_name + "."; /// A malformed loop is rejected when the stack is built; here we just stop to avoid spinning. if (!pset.insert(conf_name).second) break; } } return false; } /// Whether a non-keeper `prometheus` endpoint (serving the global `prometheus` section) /// consults the default session user. Only the time-series handlers without a fixed `user` /// authenticate; the plain metrics exposition served without a `handlers` section does not. bool prometheusHandlersConsumeDefaultSessionUser(const Poco::Util::AbstractConfiguration & config) { if (!config.has("prometheus.handlers")) return false; Poco::Util::AbstractConfiguration::Keys keys; config.keys("prometheus.handlers", keys); for (const auto & key : keys) { const std::string handler_prefix = "prometheus.handlers." + key + ".handler"; const std::string type = config.getString(handler_prefix + ".type", ""); /// The metrics exposition aliases accepted by `parseHandlerType`. if (type == "prometheus" || type == "metrics" || type == "expose_metrics" || type == "prometheus_metrics" || type == "prometheus_expose_metrics") continue; if (!config.has(handler_prefix + ".user")) return true; } return false; } } std::unique_ptr Server::buildProtocolStackFromConfig( const Poco::Util::AbstractConfiguration & config, const ServerSettings & server_settings, const std::string & protocol, Poco::Net::HTTPServerParams::Ptr http_params, AsynchronousMetrics & async_metrics, bool & is_secure, ServerType::Type & protocol_type) { /// The default session user for the endpoint: the `default_session_user` key looked up /// from the endpoint's protocol module towards the referenced (`impl`) modules; the value /// closest to the endpoint wins. If not set anywhere in the chain, the handlers fall back /// to the `default_session_user` server setting. /// It is resolved for the whole `impl` chain up front, because a module can be both typed /// (so its handler factory is created while walking the chain) and inherit the setting from /// a module it references, which the walk reaches only later. const std::optional default_session_user = getEffectiveDefaultSessionUser(config, "protocols." + protocol); auto create_factory = [&](const std::string & type, const std::string & conf_name) -> TCPServerConnectionFactory::Ptr { if (type == "tcp") return TCPServerConnectionFactory::Ptr(new TCPHandlerFactory(*this, false, false, ProfileEvents::InterfaceNativeReceiveBytes, ProfileEvents::InterfaceNativeSendBytes, default_session_user)); if (type == "tls") #if USE_SSL return TCPServerConnectionFactory::Ptr(new TLSHandlerFactory(*this, conf_name)); #else throw Exception(ErrorCodes::SUPPORT_IS_DISABLED, "SSL support for TCP protocol is disabled because Poco library was built without NetSSL support."); #endif if (type == "proxy1") return TCPServerConnectionFactory::Ptr(new ProxyV1HandlerFactory(*this, conf_name)); if (type == "mysql") return TCPServerConnectionFactory::Ptr(new MySQLHandlerFactory(*this, server_settings[ServerSetting::mysql_require_secure_transport], ProfileEvents::InterfaceMySQLReceiveBytes, ProfileEvents::InterfaceMySQLSendBytes, default_session_user)); if (type == "postgres") #if USE_SSL return TCPServerConnectionFactory::Ptr(new PostgreSQLHandlerFactory(*this, server_settings[ServerSetting::postgresql_require_secure_transport], conf_name + ".", ProfileEvents::InterfacePostgreSQLReceiveBytes, ProfileEvents::InterfacePostgreSQLSendBytes, default_session_user)); #else return TCPServerConnectionFactory::Ptr(new PostgreSQLHandlerFactory(*this, server_settings[ServerSetting::postgresql_require_secure_transport], ProfileEvents::InterfacePostgreSQLReceiveBytes, ProfileEvents::InterfacePostgreSQLSendBytes, default_session_user)); #endif if (type == "http") { /// Allow custom http_handlers configuration for this protocol endpoint. /// E.g. httphttp_handlers_alt String handlers_config_key; if (config.has(conf_name + ".handlers")) handlers_config_key = config.getString(conf_name + ".handlers"); return TCPServerConnectionFactory::Ptr( new HTTPServerConnectionFactory(httpContext(), http_params, createHandlerFactory(*this, config, async_metrics, "HTTPHandler-factory", handlers_config_key, protocol, default_session_user), ProfileEvents::InterfaceHTTPReceiveBytes, ProfileEvents::InterfaceHTTPSendBytes) ); } if (type == "prometheus") { const std::string handler_name = server_settings[ServerSetting::prometheus_keeper_metrics_only] ? "KeeperPrometheusHandler-factory" : "PrometheusHandler-factory"; return TCPServerConnectionFactory::Ptr( new HTTPServerConnectionFactory(httpContext(), http_params, createHandlerFactory(*this, config, async_metrics, handler_name, /* http_handlers_key= */ {}, protocol, default_session_user), ProfileEvents::InterfacePrometheusReceiveBytes, ProfileEvents::InterfacePrometheusSendBytes) ); } if (type == "interserver") return TCPServerConnectionFactory::Ptr( new HTTPServerConnectionFactory(httpContext(), http_params, createHandlerFactory(*this, config, async_metrics, "InterserverIOHTTPHandler-factory"), ProfileEvents::InterfaceInterserverReceiveBytes, ProfileEvents::InterfaceInterserverSendBytes) ); throw Exception(ErrorCodes::INVALID_CONFIG_PARAMETER, "Protocol configuration error, unknown protocol name '{}'", type); }; std::string conf_name = "protocols." + protocol; std::string prefix = conf_name + "."; std::unordered_set pset {conf_name}; const bool is_introspection = isIntrospectionProtocol(config, protocol); std::string innermost_type; auto stack = std::make_unique(*this, conf_name, is_introspection); bool has_interserver = false; /// `tls` and `proxy1` only wrap another protocol; the innermost layer of the chain is /// what the server actually speaks, and it determines which metrics are reported for it. std::string leaf_type; bool has_proxy = false; while (true) { // if there is no "type" - it's a reference to another protocol and this is just an endpoint if (config.has(prefix + "type")) { std::string type = config.getString(prefix + "type"); if (type == "tls") { if (is_secure) throw Exception(ErrorCodes::INVALID_CONFIG_PARAMETER, "Protocol '{}' contains more than one TLS layer", protocol); is_secure = true; } else if (type == "proxy1") has_proxy = true; else leaf_type = type; if (type == "interserver") has_interserver = true; if ((type == "tls" || type == "postgres") && !Poco::trim(config.getString(prefix + "cipherSuites", "")).empty()) { const auto private_key_file = config.getString(prefix + "privateKeyFile", ""); if (private_key_file.empty() || config.getString(prefix + "certificateFile", private_key_file).empty()) { /// builds a context of its own for an ACME certificate too, but it reads the /// layer-local TLS options only together with a key pair, so the value would be dropped. if (type == "tls" && config.has("acme")) throw Exception( ErrorCodes::INVALID_CONFIG_PARAMETER, "Protocol '{}': 'cipherSuites' in '{}' is not applied to a layer served with an ACME certificate; " "set it in the 'openSSL.server' section or give the layer its own 'privateKeyFile'", protocol, conf_name); throw Exception( ErrorCodes::INVALID_CONFIG_PARAMETER, "Protocol '{}': 'cipherSuites' in '{}' requires a 'privateKeyFile' (and 'certificateFile', if separate) " "in the same section, without them it cannot be applied", protocol, conf_name); } } if (is_introspection && type != "tcp" && type != "tls" && type != "proxy1") throw Exception( ErrorCodes::INVALID_CONFIG_PARAMETER, "Introspection protocol '{}' contains a '{}' layer, but only 'tcp' optionally wrapped " "into 'tls' and 'proxy1' layers is supported", protocol, type); innermost_type = type; stack->append(create_factory(type, conf_name)); } if (!config.has(prefix + "impl")) break; conf_name = "protocols." + config.getString(prefix + "impl"); prefix = conf_name + "."; if (!pset.insert(conf_name).second) throw Exception(ErrorCodes::INVALID_CONFIG_PARAMETER, "Protocol '{}' configuration contains a loop on '{}'", protocol, conf_name); } /// Interserver connections are authenticated by the cluster secret and the initial user /// and never use the default session user, so such a configuration is an error. if (has_interserver && default_session_user) throw Exception(ErrorCodes::INVALID_CONFIG_PARAMETER, "Protocol '{}': 'default_session_user' cannot be used with the 'interserver' protocol", protocol); if (is_introspection && innermost_type != "tcp") throw Exception( ErrorCodes::INVALID_CONFIG_PARAMETER, "Introspection protocol '{}' must end in a 'tcp' layer", protocol); protocol_type = protocolStackType(leaf_type, is_secure, has_proxy); return stack; } HTTPContextPtr Server::httpContext() const { return std::make_shared(context()); } void Server::createServers( Poco::Util::AbstractConfiguration & config, const ServerSettings & server_settings, const Strings & listen_hosts, bool listen_try, Poco::ThreadPool & server_pool, AsynchronousMetrics & async_metrics, std::vector & servers, bool start_servers, const ServerType & server_type, bool only_introspection_protocols) { const Settings & settings = global_context->getSettingsRef(); #if USE_GRPC || USE_ARROWFLIGHT const auto grpc_listen_hosts = getGRPCListenHosts(listen_hosts); #endif Poco::Net::HTTPServerParams::Ptr http_params = new Poco::Net::HTTPServerParams; http_params->setTimeout(settings[Setting::http_receive_timeout]); http_params->setKeepAliveTimeout(global_context->getServerSettings()[ServerSetting::keep_alive_timeout]); http_params->setMaxKeepAliveRequests(static_cast(global_context->getServerSettings()[ServerSetting::max_keep_alive_requests])); http_params->setMaxQueued(server_settings[ServerSetting::listen_backlog]); Poco::Util::AbstractConfiguration::Keys protocols; config.keys("protocols", protocols); const TCPServerConnectionFilter::Ptr & connection_filter = new TCPServerConnectionFilter{[&]() { return !ProfileEvents::checkCPUOverload(global_context->getServerSettings()[ServerSetting::os_cpu_busy_time_threshold], global_context->getMinOSCPUWaitTimeRatioToDropConnection(), global_context->getMaxOSCPUWaitTimeRatioToDropConnection(), /*should_throw*/ false); }}; for (const auto & protocol : protocols) { const bool is_introspection = isIntrospectionProtocol(config, protocol); if (is_introspection != only_introspection_protocols) continue; if (!server_type.shouldStart(ServerType::Type::CUSTOM, protocol)) continue; std::string prefix = "protocols." + protocol + "."; std::string port_name = prefix + "port"; std::string description {" protocol"}; if (config.has(prefix + "description")) description = config.getString(prefix + "description"); if (!config.has(prefix + "port")) continue; std::vector hosts; if (config.has(prefix + "host")) hosts.push_back(config.getString(prefix + "host")); else hosts = listen_hosts; for (const auto & host : hosts) { bool is_secure = false; auto protocol_type = ServerType::Type::END; auto stack = buildProtocolStackFromConfig( config, server_settings, protocol, http_params, async_metrics, is_secure, protocol_type); if (stack->empty()) throw Exception(ErrorCodes::INVALID_CONFIG_PARAMETER, "Protocol '{}' stack empty", protocol); createServer(config, host, port_name.c_str(), listen_try, start_servers, servers, [&](UInt16 port) -> ProtocolServerAdapter { Poco::Net::ServerSocket socket; auto address = socketBindListen(server_settings, socket, host, port, is_secure); socket.setReceiveTimeout(settings[Setting::receive_timeout]); socket.setSendTimeout(settings[Setting::send_timeout]); return ProtocolServerAdapter( host, port_name.c_str(), protocol_type, description + ": " + address.toString(), std::make_unique( stack.release(), server_pool, socket, makeServerParams(server_settings), is_introspection ? TCPServerConnectionFilter::Ptr() : connection_filter)); }); } } /// Introspection ports can only be configured in the protocols config section. if (only_introspection_protocols) return; for (const auto & listen_host : listen_hosts) { const char * port_name = nullptr; if (server_type.shouldStart(ServerType::Type::HTTP)) { /// HTTP port_name = "http_port"; /// Handler configuration is parsed outside the callback: anything the callback throws is /// reported as a listener bind failure, and only logged when `listen_try` is set. The port /// check matches the early return in `createServer`. if (!config.getString(port_name, "").empty()) { auto handler_factory = createHandlerFactory(*this, config, async_metrics, "HTTPHandler-factory"); createServer(config, listen_host, port_name, listen_try, start_servers, servers, [&](UInt16 port) -> ProtocolServerAdapter { Poco::Net::ServerSocket socket; auto address = socketBindListen(server_settings, socket, listen_host, port); socket.setReceiveTimeout(settings[Setting::http_receive_timeout]); socket.setSendTimeout(settings[Setting::http_send_timeout]); return ProtocolServerAdapter( listen_host, port_name, ServerType::Type::HTTP, "http://" + address.toString(), std::make_unique( httpContext(), handler_factory, server_pool, socket, http_params, connection_filter, ProfileEvents::InterfaceHTTPReceiveBytes, ProfileEvents::InterfaceHTTPSendBytes)); }); } } if (server_type.shouldStart(ServerType::Type::HTTPS)) { /// HTTPS port_name = "https_port"; if (!config.getString(port_name, "").empty()) { #if USE_SSL auto handler_factory = createHandlerFactory(*this, config, async_metrics, "HTTPSHandler-factory"); #endif createServer(config, listen_host, port_name, listen_try, start_servers, servers, [&](UInt16 port) -> ProtocolServerAdapter { #if USE_SSL Poco::Net::SecureServerSocket socket; auto address = socketBindListen(server_settings, socket, listen_host, port, /* secure = */ true); socket.setReceiveTimeout(settings[Setting::http_receive_timeout]); socket.setSendTimeout(settings[Setting::http_send_timeout]); return ProtocolServerAdapter( listen_host, port_name, ServerType::Type::HTTPS, "https://" + address.toString(), std::make_unique( httpContext(), handler_factory, server_pool, socket, http_params, connection_filter, ProfileEvents::InterfaceHTTPReceiveBytes, ProfileEvents::InterfaceHTTPSendBytes)); #else UNUSED(port); throw Exception(ErrorCodes::SUPPORT_IS_DISABLED, "HTTPS protocol is disabled because Poco library was built without NetSSL support."); #endif }); } } if (server_type.shouldStart(ServerType::Type::TCP)) { /// TCP port_name = "tcp_port"; createServer(config, listen_host, port_name, listen_try, start_servers, servers, [&](UInt16 port) -> ProtocolServerAdapter { Poco::Net::ServerSocket socket; auto address = socketBindListen(server_settings, socket, listen_host, port); socket.setReceiveTimeout(settings[Setting::receive_timeout]); socket.setSendTimeout(settings[Setting::send_timeout]); return ProtocolServerAdapter( listen_host, port_name, ServerType::Type::TCP, "native protocol (tcp): " + address.toString(), std::make_unique( new TCPHandlerFactory(*this, /* secure */ false, /* proxy protocol */ false, ProfileEvents::InterfaceNativeReceiveBytes, ProfileEvents::InterfaceNativeSendBytes), server_pool, socket, makeServerParams(server_settings), connection_filter)); }); } if (server_type.shouldStart(ServerType::Type::TCP_WITH_PROXY)) { /// TCP with PROXY protocol, see https://github.com/wolfeidau/proxyv2/blob/master/docs/proxy-protocol.txt port_name = "tcp_with_proxy_port"; createServer(config, listen_host, port_name, listen_try, start_servers, servers, [&](UInt16 port) -> ProtocolServerAdapter { Poco::Net::ServerSocket socket; auto address = socketBindListen(server_settings, socket, listen_host, port); socket.setReceiveTimeout(settings[Setting::receive_timeout]); socket.setSendTimeout(settings[Setting::send_timeout]); return ProtocolServerAdapter( listen_host, port_name, ServerType::Type::TCP_WITH_PROXY, "native protocol (tcp) with PROXY: " + address.toString(), std::make_unique( new TCPHandlerFactory(*this, /* secure */ false, /* proxy protocol */ true, ProfileEvents::InterfaceNativeReceiveBytes, ProfileEvents::InterfaceNativeSendBytes), server_pool, socket, makeServerParams(server_settings), connection_filter)); }); } #if USE_ARROWFLIGHT if (server_type.shouldStart(ServerType::Type::ARROW_FLIGHT) && std::ranges::find(grpc_listen_hosts, listen_host) != grpc_listen_hosts.end()) { port_name = "arrowflight_port"; createServer(config, listen_host, port_name, listen_try, start_servers, servers, [&](UInt16 port) -> ProtocolServerAdapter { /// Do not bind a Poco socket here: gRPC owns the listening socket of an Arrow Flight /// server and binds it in `start`. A pre-bind would both bind the address twice and /// defeat the family-agnostic handling of a wildcard address by gRPC, which falls /// back to `0.0.0.0` for `::` on a host without IPv6. auto address = makeSocketAddress(listen_host, port, &logger()); return ProtocolServerAdapter( listen_host, port_name, ServerType::Type::ARROW_FLIGHT, "Arrow Flight compatibility protocol: " + address.toString(), std::unique_ptr(new ArrowFlightServer(*this, address)), true); }); } #endif if (server_type.shouldStart(ServerType::Type::TCP_SECURE)) { /// TCP with SSL port_name = "tcp_port_secure"; createServer(config, listen_host, port_name, listen_try, start_servers, servers, [&](UInt16 port) -> ProtocolServerAdapter { #if USE_SSL Poco::Net::SecureServerSocket socket; auto address = socketBindListen(server_settings, socket, listen_host, port, /* secure = */ true); socket.setReceiveTimeout(settings[Setting::receive_timeout]); socket.setSendTimeout(settings[Setting::send_timeout]); return ProtocolServerAdapter( listen_host, port_name, ServerType::Type::TCP_SECURE, "secure native protocol (tcp_secure): " + address.toString(), std::make_unique( new TCPHandlerFactory(*this, /* secure */ true, /* proxy protocol */ false, ProfileEvents::InterfaceNativeReceiveBytes, ProfileEvents::InterfaceNativeSendBytes), server_pool, socket, makeServerParams(server_settings), connection_filter)); #else UNUSED(port); throw Exception(ErrorCodes::SUPPORT_IS_DISABLED, "SSL support for TCP protocol is disabled because Poco library was built without NetSSL support."); #endif }); } if (server_type.shouldStart(ServerType::Type::TCP_SSH)) { port_name = "tcp_ssh_port"; createServer( config, listen_host, port_name, listen_try, start_servers, servers, [&](UInt16 port) -> ProtocolServerAdapter { #if USE_SSH && defined(OS_LINUX) Poco::Net::ServerSocket socket; auto address = socketBindListen(server_settings, socket, listen_host, port, /* secure = */ false); return ProtocolServerAdapter( listen_host, port_name, ServerType::Type::TCP_SSH, "SSH PTY: " + address.toString(), std::make_unique( new SSHPtyHandlerFactory(*this, config), server_pool, socket, makeServerParams(server_settings), connection_filter)); #else UNUSED(port); throw Exception(ErrorCodes::SUPPORT_IS_DISABLED, "SSH protocol is disabled because ClickHouse has been built without libssh or is running on a non-Linux platform"); #endif }); } if (server_type.shouldStart(ServerType::Type::MYSQL)) { port_name = "mysql_port"; createServer(config, listen_host, port_name, listen_try, start_servers, servers, [&](UInt16 port) -> ProtocolServerAdapter { Poco::Net::ServerSocket socket; auto address = socketBindListen(server_settings, socket, listen_host, port, /* secure = */ true); socket.setReceiveTimeout(Poco::Timespan()); socket.setSendTimeout(settings[Setting::send_timeout]); bool secure_required = server_settings[ServerSetting::mysql_require_secure_transport]; return ProtocolServerAdapter( listen_host, port_name, ServerType::Type::MYSQL, "MySQL compatibility protocol: " + address.toString(), std::make_unique( new MySQLHandlerFactory(*this, secure_required, ProfileEvents::InterfaceMySQLReceiveBytes, ProfileEvents::InterfaceMySQLSendBytes), server_pool, socket, makeServerParams(server_settings), connection_filter)); }); } if (server_type.shouldStart(ServerType::Type::POSTGRESQL)) { port_name = "postgresql_port"; createServer(config, listen_host, port_name, listen_try, start_servers, servers, [&](UInt16 port) -> ProtocolServerAdapter { Poco::Net::ServerSocket socket; auto address = socketBindListen(server_settings, socket, listen_host, port, /* secure = */ true); socket.setReceiveTimeout(Poco::Timespan()); socket.setSendTimeout(settings[Setting::send_timeout]); bool secure_required = server_settings[ServerSetting::postgresql_require_secure_transport]; return ProtocolServerAdapter( listen_host, port_name, ServerType::Type::POSTGRESQL, "PostgreSQL compatibility protocol: " + address.toString(), #if USE_SSL std::make_unique( new PostgreSQLHandlerFactory(*this, secure_required, Poco::Net::SSLManager::CFG_SERVER_PREFIX, ProfileEvents::InterfacePostgreSQLReceiveBytes, ProfileEvents::InterfacePostgreSQLSendBytes), server_pool, socket, makeServerParams(server_settings), connection_filter)); #else std::make_unique( new PostgreSQLHandlerFactory(*this, secure_required, ProfileEvents::InterfacePostgreSQLReceiveBytes, ProfileEvents::InterfacePostgreSQLSendBytes), server_pool, socket, makeServerParams(server_settings), connection_filter)); #endif }); } #if USE_GRPC if (server_type.shouldStart(ServerType::Type::GRPC) && std::ranges::find(grpc_listen_hosts, listen_host) != grpc_listen_hosts.end()) { port_name = "grpc_port"; createServer(config, listen_host, port_name, listen_try, start_servers, servers, [&](UInt16 port) -> ProtocolServerAdapter { auto server_address = makeSocketAddress(listen_host, port, &logger()); return ProtocolServerAdapter( listen_host, port_name, ServerType::Type::GRPC, "gRPC protocol: " + server_address.toString(), std::make_unique(*this, server_address)); }); } #endif if (server_type.shouldStart(ServerType::Type::PROMETHEUS)) { /// Prometheus (if defined and not setup yet with http_port) port_name = "prometheus.port"; const char * handler_name = server_settings[ServerSetting::prometheus_keeper_metrics_only] ? "KeeperPrometheusHandler-factory" : "PrometheusHandler-factory"; if (!config.getString(port_name, "").empty()) { auto handler_factory = createHandlerFactory(*this, config, async_metrics, handler_name); createServer(config, listen_host, port_name, listen_try, start_servers, servers, [&](UInt16 port) -> ProtocolServerAdapter { Poco::Net::ServerSocket socket; auto address = socketBindListen(server_settings, socket, listen_host, port); socket.setReceiveTimeout(settings[Setting::http_receive_timeout]); socket.setSendTimeout(settings[Setting::http_send_timeout]); return ProtocolServerAdapter( listen_host, port_name, ServerType::Type::PROMETHEUS, "Prometheus: http://" + address.toString(), std::make_unique( httpContext(), handler_factory, server_pool, socket, http_params, nullptr, ProfileEvents::InterfacePrometheusReceiveBytes, ProfileEvents::InterfacePrometheusSendBytes)); }); } } bool has_catalog_port = !config.getString("iceberg_rest_catalog.port", "").empty(); bool has_catalog_secure_port = !config.getString("iceberg_rest_catalog.secure_port", "").empty(); if (server_type.shouldStart(ServerType::Type::ICEBERG_REST_CATALOG) && (has_catalog_port || has_catalog_secure_port)) { auto is_listening = [&](const char * name) { return std::any_of(servers.begin(), servers.end(), [&](const auto & server) { return !server.isStopping() && server.getListenHost() == listen_host && server.getPortName() == name; }); }; /// Both listeners share one handler factory. Create it only if some configured port is not listening yet. const bool already_listening = (!has_catalog_port || is_listening("iceberg_rest_catalog.port")) && (!has_catalog_secure_port || is_listening("iceberg_rest_catalog.secure_port")); HTTPRequestHandlerFactoryPtr handler_factory; if (!already_listening) { try { handler_factory = createIcebergRESTCatalogHandlerFactory(*this, config); } catch (...) { LOG_ERROR(&logger(), "Not starting the Iceberg REST catalog server: {}", getCurrentExceptionMessage(/*with_stacktrace*/ false)); } } if (handler_factory) { if (has_catalog_port) { port_name = "iceberg_rest_catalog.port"; createServer(config, listen_host, port_name, listen_try, start_servers, servers, [&](UInt16 port) -> ProtocolServerAdapter { Poco::Net::ServerSocket socket; auto address = socketBindListen(server_settings, socket, listen_host, port); socket.setReceiveTimeout(settings[Setting::http_receive_timeout]); socket.setSendTimeout(settings[Setting::http_send_timeout]); return ProtocolServerAdapter( listen_host, port_name, ServerType::Type::ICEBERG_REST_CATALOG, "Iceberg REST catalog: http://" + address.toString(), std::make_unique( httpContext(), handler_factory, server_pool, socket, http_params, nullptr, ProfileEvents::InterfaceHTTPReceiveBytes, ProfileEvents::InterfaceHTTPSendBytes)); }); } /// The TLS listener uses the certificate from the section. if (has_catalog_secure_port) { port_name = "iceberg_rest_catalog.secure_port"; createServer(config, listen_host, port_name, listen_try, start_servers, servers, [&]([[maybe_unused]] UInt16 port) -> ProtocolServerAdapter { #if USE_SSL Poco::Net::SecureServerSocket socket; auto address = socketBindListen(server_settings, socket, listen_host, port, /* secure = */ true); socket.setReceiveTimeout(settings[Setting::http_receive_timeout]); socket.setSendTimeout(settings[Setting::http_send_timeout]); return ProtocolServerAdapter( listen_host, port_name, ServerType::Type::ICEBERG_REST_CATALOG, "Iceberg REST catalog: https://" + address.toString(), std::make_unique( httpContext(), handler_factory, server_pool, socket, http_params, nullptr, ProfileEvents::InterfaceHTTPReceiveBytes, ProfileEvents::InterfaceHTTPSendBytes)); #else throw Exception(ErrorCodes::SUPPORT_IS_DISABLED, "HTTPS for the Iceberg REST catalog is disabled because Poco library was built without NetSSL support."); #endif }); } } } } } void Server::createInterserverServers( Poco::Util::AbstractConfiguration & config, const ServerSettings & server_settings, const Strings & interserver_listen_hosts, bool listen_try, Poco::ThreadPool & server_pool, AsynchronousMetrics & async_metrics, std::vector & servers, bool start_servers, const ServerType & server_type) { const Settings & settings = global_context->getSettingsRef(); Poco::Net::HTTPServerParams::Ptr http_params = new Poco::Net::HTTPServerParams; http_params->setTimeout(settings[Setting::http_receive_timeout]); http_params->setKeepAliveTimeout(global_context->getServerSettings()[ServerSetting::keep_alive_timeout]); http_params->setMaxQueued(server_settings[ServerSetting::listen_backlog]); /// Now iterate over interserver_listen_hosts for (const auto & interserver_listen_host : interserver_listen_hosts) { const char * port_name = nullptr; if (server_type.shouldStart(ServerType::Type::INTERSERVER_HTTP)) { /// Interserver IO HTTP port_name = "interserver_http_port"; createServer(config, interserver_listen_host, port_name, listen_try, start_servers, servers, [&](UInt16 port) -> ProtocolServerAdapter { Poco::Net::ServerSocket socket; auto address = socketBindListen(server_settings, socket, interserver_listen_host, port); socket.setReceiveTimeout(settings[Setting::http_receive_timeout]); socket.setSendTimeout(settings[Setting::http_send_timeout]); return ProtocolServerAdapter( interserver_listen_host, port_name, ServerType::Type::INTERSERVER_HTTP, "replica communication (interserver): http://" + address.toString(), std::make_unique( httpContext(), createHandlerFactory(*this, config, async_metrics, "InterserverIOHTTPHandler-factory"), server_pool, socket, http_params, nullptr, ProfileEvents::InterfaceInterserverReceiveBytes, ProfileEvents::InterfaceInterserverSendBytes)); }); } if (server_type.shouldStart(ServerType::Type::INTERSERVER_HTTPS)) { port_name = "interserver_https_port"; createServer(config, interserver_listen_host, port_name, listen_try, start_servers, servers, [&](UInt16 port) -> ProtocolServerAdapter { #if USE_SSL Poco::Net::SecureServerSocket socket; auto address = socketBindListen(server_settings, socket, interserver_listen_host, port, /* secure = */ true); socket.setReceiveTimeout(settings[Setting::http_receive_timeout]); socket.setSendTimeout(settings[Setting::http_send_timeout]); return ProtocolServerAdapter( interserver_listen_host, port_name, ServerType::Type::INTERSERVER_HTTPS, "secure replica communication (interserver): https://" + address.toString(), std::make_unique( httpContext(), createHandlerFactory(*this, config, async_metrics, "InterserverIOHTTPSHandler-factory"), server_pool, socket, http_params, nullptr, ProfileEvents::InterfaceInterserverReceiveBytes, ProfileEvents::InterfaceInterserverSendBytes)); #else UNUSED(port); throw Exception(ErrorCodes::SUPPORT_IS_DISABLED, "SSL support for TCP protocol is disabled because Poco library was built without NetSSL support."); #endif }); } } } void Server::stopServers( std::vector & servers, const ServerType & server_type) const { DB::stopServers(servers, server_type, &logger()); } void Server::updateServers( Poco::Util::AbstractConfiguration & config, const ServerSettings & server_settings, Poco::ThreadPool & server_pool, AsynchronousMetrics & async_metrics, std::vector & servers, std::vector & servers_to_start_before_tables) { LoggerRawPtr log = &logger(); const auto listen_hosts = getListenHosts(config); const auto interserver_listen_hosts = getInterserverListenHosts(config); const auto listen_try = getListenTry(config, server_settings); #if USE_GRPC || USE_ARROWFLIGHT const auto grpc_listen_hosts = getGRPCListenHosts(listen_hosts); #endif /// Remove servers once all their connections are closed auto check_server = [&log](const char prefix[], auto & server) { if (!server.isStopping()) return false; size_t current_connections = server.currentConnections(); LOG_DEBUG(log, "Server {}{}: {} ({} connections)", server.getDescription(), prefix, !current_connections ? "finished" : "waiting", current_connections); return !current_connections; }; std::erase_if(servers, std::bind_front(check_server, " (from one of previous reload)")); Poco::Util::AbstractConfiguration & previous_config = latest_config ? *latest_config : config; std::vector all_servers; all_servers.reserve(servers.size() + servers_to_start_before_tables.size()); for (auto & server : servers) all_servers.push_back(&server); for (auto & server : servers_to_start_before_tables) all_servers.push_back(&server); for (auto * server : all_servers) { if (server->supportsRuntimeReconfiguration() && !server->isStopping()) { std::string port_name = server->getPortName(); bool has_host = false; bool is_http = false; bool is_non_keeper_prometheus = false; bool is_prometheus = false; bool default_session_user_changed = false; String handlers_key = "http_handlers"; String previous_handlers_key = handlers_key; if (port_name.starts_with("protocols.")) { std::string protocol = port_name.substr(0, port_name.find_last_of('.')); has_host = config.has(protocol + ".host"); /// Whether any handler in the `impl` chain actually consumes the default session /// user, so that a `default_session_user` change does not restart listeners that /// ignore it: keeper-metrics-only `prometheus` listeners serve metrics without /// authentication (`KeeperPrometheusHandler-factory` only exposes `MetricsImpl`), /// and `prometheus` endpoints whose configured handler set consists only of /// fixed-user and non-authenticating handlers never consult the setting either. bool consumes_default_session_user = false; std::string conf_name = protocol; std::string prefix = protocol + "."; std::unordered_set pset {conf_name}; while (true) { if (config.has(prefix + "type")) { std::string type = config.getString(prefix + "type"); if (type == "http") { is_http = true; if (config.has(conf_name + ".handlers")) handlers_key = config.getString(conf_name + ".handlers"); /// This reload may re-point the endpoint at a different `` section /// (through the endpoint's `handlers` reference or its `impl` chain), while /// the running factory still serves the previously referenced section. Resolve /// the old reference too: the consumer check must consider the old *or* the /// new handler set, and the changed-configuration check below must notice the /// reference switch itself — comparing only the new section on both sides of /// the reload would miss a switch between two sections that are themselves /// unchanged. previous_handlers_key = resolveHTTPHandlersKey(previous_config, protocol).value_or(handlers_key); /// `SQLDefinedHTTPHandlerFactory` is appended to every HTTP handler /// factory. It caches the endpoint override, while the SQL handler /// registry itself is read on every request, so an SQL-defined handler /// can start consuming the setting after the listener was created even /// when all XML handlers have fixed users. consumes_default_session_user = true; break; } if (type == "tcp" || type == "mysql" || type == "postgres") consumes_default_session_user = true; /// The running listener still serves the handler set of the *previous* /// configuration (the shared `prometheus.handlers` section), so consider the /// setting consumed when either the old or the new set consumes it. A change /// of the section itself forces a restart below, like a change of an `http` /// endpoint's `handlers` section. if (type == "prometheus") { is_prometheus = true; if (!server_settings[ServerSetting::prometheus_keeper_metrics_only]) { is_non_keeper_prometheus = true; consumes_default_session_user = consumes_default_session_user || prometheusHandlersConsumeDefaultSessionUser(previous_config) || prometheusHandlersConsumeDefaultSessionUser(config); } } } if (!config.has(prefix + "impl")) break; conf_name = "protocols." + config.getString(prefix + "impl"); prefix = conf_name + "."; if (!pset.insert(conf_name).second) throw Exception(ErrorCodes::INVALID_CONFIG_PARAMETER, "Protocol '{}' configuration contains a loop on '{}'", protocol, conf_name); } /// The per-endpoint default session user is fixed in the protocol handler factory, /// so the endpoint must be restarted when its effective value changes. Compare the /// effective value (the one closest to the endpoint wins, as in /// `buildProtocolStackFromConfig`) rather than every node in the `impl` chain, so /// that editing a shadowed base module does not force a needless restart. default_session_user_changed = consumes_default_session_user && getEffectiveDefaultSessionUser(previous_config, protocol) != getEffectiveDefaultSessionUser(config, protocol); } else { /// NOTE: better to compare using getPortName() over using /// dynamic_cast<> since HTTPServer is also used for prometheus and /// internal replication communications. is_http = server->getPortName() == "http_port" || server->getPortName() == "https_port"; is_prometheus = server->getPortName() == "prometheus.port"; /// The standalone `prometheus.port` listener is built from the same shared /// `prometheus.handlers` section as a composable `type = prometheus` endpoint /// (`createPrometheusMainHandlerFactory` bakes it into the handler factory), and /// there is no other reconfiguration hook for it, so it needs the same restart /// check. In keeper-metrics-only mode the section is not served at all, and a /// change of the mode itself is handled by the check below. is_non_keeper_prometheus = is_prometheus && !server_settings[ServerSetting::prometheus_keeper_metrics_only]; } if (!has_host) { const auto & server_listen_hosts = #if USE_GRPC || USE_ARROWFLIGHT (port_name == "grpc_port" || port_name == "arrowflight_port") ? grpc_listen_hosts : #endif listen_hosts; has_host = std::find(server_listen_hosts.begin(), server_listen_hosts.end(), server->getListenHost()) != server_listen_hosts.end(); } bool has_port = !config.getString(port_name, "").empty(); bool force_restart = false; if (is_http && previous_handlers_key != handlers_key) { force_restart = true; LOG_TRACE(log, "The handlers reference had been changed from <{}> to <{}>, will reload {}", previous_handlers_key, handlers_key, server->getDescription()); } else if (is_http && !isSameConfiguration(previous_config, config, handlers_key)) { force_restart = true; LOG_TRACE(log, "<{}> had been changed, will reload {}", handlers_key, server->getDescription()); } /// A non-keeper `prometheus` listener (a composable `type = prometheus` endpoint as well /// as the standalone `prometheus.port` one) serves the shared `prometheus.handlers` /// section, which is baked into its handler factory, so the listener must be restarted /// when the section changes (a fixed `user`, a route, or a handler type may have changed). if (is_non_keeper_prometheus && !isSameConfiguration(previous_config, config, "prometheus.handlers")) { force_restart = true; LOG_TRACE(log, " had been changed, will reload {}", server->getDescription()); } /// `prometheus.keeper_metrics_only` selects the handler factory baked into a `prometheus` /// listener: `KeeperPrometheusHandler-factory` exposes the keeper metrics without /// authentication, while `PrometheusHandler-factory` also serves the authenticating /// time-series handlers of `prometheus.handlers`. A change of the mode itself must restart /// the listener in either direction — the port and the handler section may well stay the /// same, and neither of the checks above can notice the switched factory then. if (is_prometheus && previous_config.getBool("prometheus.keeper_metrics_only", false) != server_settings[ServerSetting::prometheus_keeper_metrics_only]) { force_restart = true; LOG_TRACE(log, " had been changed, will reload {}", server->getDescription()); } /// The warehouse name is baked into the Iceberg REST catalog handler factory, so if /// the section changes, the listener must be restarted. if ((port_name == "iceberg_rest_catalog.port" || port_name == "iceberg_rest_catalog.secure_port") && !isSameConfiguration(previous_config, config, "iceberg_rest_catalog")) { force_restart = true; LOG_TRACE(log, " had been changed, will reload {}", server->getDescription()); } /// `asynchronous_metrics_key_values_mode` decides whether the keys of the key-value asynchronous /// metrics are written as Prometheus labels (`device="sda"`) or mangled into the metric name. A /// listener that exposes metrics is built with the constant labels and the form of its /// configuration, and neither is re-read while it runs, so it has to be rebuilt for the new form /// to be published with a label set that matches it. Only listeners that can actually serve the /// metrics protocol are rebuilt: a keeper-metrics-only `prometheus` listener exposes no /// asynchronous metrics at all, and an HTTP listener does so only through a rule with a /// `prometheus` handler type or through the default `/metrics` route - a change of the mode must /// not stop the HTTP interface of a server that only answers queries over it. The old and the new /// handler set are both consulted, because a rule may have just been added or removed. if (getAsynchronousMetricsKeyValuesMode(previous_config) != getAsynchronousMetricsKeyValuesMode(config) && (is_non_keeper_prometheus || (is_http && (httpHandlersCanExposePrometheusMetrics(previous_config, previous_handlers_key) || httpHandlersCanExposePrometheusMetrics(config, handlers_key))))) { force_restart = true; LOG_TRACE(log, " had been changed, will reload {}", server->getDescription()); } if (default_session_user_changed) { force_restart = true; LOG_TRACE(log, " had been changed, will reload {}", server->getDescription()); } if (!has_host || !has_port || config.getInt(server->getPortName()) != server->portNumber() || force_restart) { server->stop(); LOG_INFO(log, "Stopped listening for {}", server->getDescription()); } } } createServers(config, server_settings, listen_hosts, listen_try, server_pool, async_metrics, servers, /* start_servers= */ true); createInterserverServers(config, server_settings, interserver_listen_hosts, listen_try, server_pool, async_metrics, servers_to_start_before_tables, /* start_servers= */ true); std::erase_if(servers, std::bind_front(check_server, "")); std::erase_if(servers_to_start_before_tables, std::bind_front(check_server, "")); } }