From beae240077376ea5d1788033bb5b291fc4187141 Mon Sep 17 00:00:00 2001 From: guruprasad Date: Tue, 20 Oct 2026 04:16:31 +0530 Subject: [PATCH 1/2] Matches printable ASCII header values with optional internal spaces or tabs. --- .../api/internal/HttpAPIClientHelper.java | 2715 ++++++++--------- 1 file changed, 1357 insertions(+), 1358 deletions(-) diff --git a/client-v2/src/main/java/com/clickhouse/client/api/internal/HttpAPIClientHelper.java b/client-v2/src/main/java/com/clickhouse/client/api/internal/HttpAPIClientHelper.java index 70db8d438..fc942bf0d 100644 --- a/client-v2/src/main/java/com/clickhouse/client/api/internal/HttpAPIClientHelper.java +++ b/client-v2/src/main/java/com/clickhouse/client/api/internal/HttpAPIClientHelper.java @@ -1,1358 +1,1357 @@ -package com.clickhouse.client.api.internal; - -import com.clickhouse.client.api.ClickHouseException; -import com.clickhouse.client.api.Client; -import com.clickhouse.client.api.ClientConfigProperties; -import com.clickhouse.client.api.ClientException; -import com.clickhouse.client.api.ClientFaultCause; -import com.clickhouse.client.api.ClientMisconfigurationException; -import com.clickhouse.client.api.CompressionMethod; -import com.clickhouse.client.api.ConnectionInitiationException; -import com.clickhouse.client.api.ConnectionReuseStrategy; -import com.clickhouse.client.api.DataTransferException; -import com.clickhouse.client.api.ServerException; -import com.clickhouse.client.api.TransportException; -import com.clickhouse.client.api.enums.ProxyType; -import com.clickhouse.client.api.enums.SSLMode; -import com.clickhouse.client.api.http.ClickHouseHttpProto; -import com.clickhouse.client.api.observability.DefaultSpanRecorder; -import com.clickhouse.client.api.observability.Span; -import com.clickhouse.client.api.observability.SpanRecorder; -import com.clickhouse.client.api.transport.Endpoint; -import com.clickhouse.client.api.transport.internal.TransportRequest; -import com.clickhouse.client.api.transport.internal.TransportResponse; -import com.clickhouse.data.ClickHouseFormat; -import net.jpountz.lz4.LZ4Factory; -import org.apache.commons.compress.compressors.CompressorStreamFactory; -import org.apache.hc.client5.http.ConnectTimeoutException; -import org.apache.hc.client5.http.classic.methods.HttpPost; -import org.apache.hc.client5.http.config.ConnectionConfig; -import org.apache.hc.client5.http.config.RequestConfig; -import org.apache.hc.client5.http.entity.mime.MultipartEntityBuilder; -import org.apache.hc.client5.http.impl.classic.CloseableHttpClient; -import org.apache.hc.client5.http.impl.classic.HttpClientBuilder; -import org.apache.hc.client5.http.impl.classic.RequestFailedException; -import org.apache.hc.client5.http.impl.io.BasicHttpClientConnectionManager; -import org.apache.hc.client5.http.impl.io.ManagedHttpClientConnectionFactory; -import org.apache.hc.client5.http.impl.io.PoolingHttpClientConnectionManager; -import org.apache.hc.client5.http.impl.io.PoolingHttpClientConnectionManagerBuilder; -import org.apache.hc.client5.http.io.HttpClientConnectionManager; -import org.apache.hc.client5.http.io.ManagedHttpClientConnection; -import org.apache.hc.client5.http.protocol.HttpClientContext; -import org.apache.hc.client5.http.socket.LayeredConnectionSocketFactory; -import org.apache.hc.client5.http.ssl.SSLConnectionSocketFactory; -import org.apache.hc.client5.http.ssl.TlsSocketStrategy; -import org.apache.hc.core5.http.ClassicHttpResponse; -import org.apache.hc.core5.http.ConnectionRequestTimeoutException; -import org.apache.hc.core5.http.ContentType; -import org.apache.hc.core5.http.Header; -import org.apache.hc.core5.http.HttpEntity; -import org.apache.hc.core5.http.HttpHeaders; -import org.apache.hc.core5.http.HttpHost; -import org.apache.hc.core5.http.HttpRequest; -import org.apache.hc.core5.http.HttpResponse; -import org.apache.hc.core5.http.HttpStatus; -import org.apache.hc.core5.http.NoHttpResponseException; -import org.apache.hc.core5.http.ProtocolException; -import org.apache.hc.core5.http.URIScheme; -import org.apache.hc.core5.http.config.CharCodingConfig; -import org.apache.hc.core5.http.config.Http1Config; -import org.apache.hc.core5.http.config.Lookup; -import org.apache.hc.core5.http.config.RegistryBuilder; -import org.apache.hc.core5.http.impl.io.DefaultHttpResponseParserFactory; -import org.apache.hc.core5.http.io.SocketConfig; -import org.apache.hc.core5.http.io.entity.ByteArrayEntity; -import org.apache.hc.core5.http.io.entity.EntityTemplate; -import org.apache.hc.core5.http.protocol.HttpContext; -import org.apache.hc.core5.io.CloseMode; -import org.apache.hc.core5.io.IOCallback; -import org.apache.hc.core5.net.URIAuthority; -import org.apache.hc.core5.net.URIBuilder; -import org.apache.hc.core5.pool.ConnPoolControl; -import org.apache.hc.core5.pool.PoolConcurrencyPolicy; -import org.apache.hc.core5.pool.PoolReusePolicy; -import org.apache.hc.core5.util.TimeValue; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import javax.net.ssl.HostnameVerifier; -import javax.net.ssl.SNIHostName; -import javax.net.ssl.SSLContext; -import javax.net.ssl.SSLException; -import javax.net.ssl.SSLParameters; -import javax.net.ssl.SSLSocket; -import java.io.IOException; -import java.io.InputStream; -import java.io.OutputStream; -import java.io.UnsupportedEncodingException; -import java.lang.reflect.Method; -import java.net.ConnectException; -import java.net.InetSocketAddress; -import java.net.NoRouteToHostException; -import java.net.Socket; -import java.net.SocketTimeoutException; -import java.net.URI; -import java.net.URISyntaxException; -import java.net.URLEncoder; -import java.net.UnknownHostException; -import java.nio.charset.StandardCharsets; -import java.util.Arrays; -import java.util.Base64; -import java.util.Collection; -import java.util.Collections; -import java.util.HashMap; -import java.util.HashSet; -import java.util.List; -import java.util.Map; -import java.util.Objects; -import java.util.Properties; -import java.util.Set; -import java.util.concurrent.ConcurrentLinkedQueue; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicBoolean; -import java.util.concurrent.atomic.AtomicLong; -import java.util.function.BiConsumer; -import java.util.function.Function; -import java.util.regex.Pattern; -import java.util.stream.Stream; - -public class HttpAPIClientHelper { - - public static final String KEY_STATEMENT_PARAMS = "statement_params"; - - private static final Logger LOG = LoggerFactory.getLogger(HttpAPIClientHelper.class); - - private static final int ERROR_BODY_BUFFER_SIZE = 1024; // Error messages are usually small - - private final String DEFAULT_HTTP_COMPRESSION_ALGO = "lz4"; - - private static final Pattern PATTERN_HEADER_VALUE_ASCII = Pattern.compile( - "\\p{Graph}+(?:[ ]\\p{Graph}+)*"); - - private final CloseableHttpClient httpClient; - - private String proxyAuthHeaderValue; - - private final Set defaultRetryCauses; - - private final String defaultUserAgent; - private final Object metricsRegistry; - - ConnPoolControl poolControl; - - LZ4Factory lz4Factory; - - private final SslContextProvider sslContextProvider = new SslContextProvider(); - - /** - * Recorder a span is started on per transport request. Never {@code null} - - * {@link com.clickhouse.client.api.observability.DefaultSpanRecorder#NOOP} when observability is - * not configured. - */ - private final SpanRecorder spanRecorder; - - public HttpAPIClientHelper(Map configuration, Object metricsRegistry, boolean initSslContext, - LZ4Factory lz4Factory, SpanRecorder spanRecorder) { - this.spanRecorder = Objects.requireNonNull(spanRecorder, - "spanRecorder is required; use DefaultSpanRecorder.NOOP to record nothing"); - this.metricsRegistry = metricsRegistry; - this.httpClient = createHttpClient(initSslContext, configuration); - this.lz4Factory = lz4Factory; - assert this.lz4Factory != null; - - boolean usingClientCompression = ClientConfigProperties.COMPRESS_CLIENT_REQUEST.getOrDefault(configuration); - boolean usingServerCompression = ClientConfigProperties.COMPRESS_SERVER_RESPONSE.getOrDefault(configuration); - boolean useHttpCompression = ClientConfigProperties.USE_HTTP_COMPRESSION.getOrDefault(configuration); - - LOG.debug("client compression: {}, server compression: {}, http compression: {}", usingClientCompression, usingServerCompression, useHttpCompression); - - defaultRetryCauses = new HashSet<>(ClientConfigProperties.CLIENT_RETRY_ON_FAILURE.getOrDefault(configuration)); - if (defaultRetryCauses.contains(ClientFaultCause.None)) { - defaultRetryCauses.removeIf(c -> c != ClientFaultCause.None); - } - - this.defaultUserAgent = buildDefaultUserAgent(); - } - - /** - * Creates an SSL context from the configured trust/key material. - * - * @return SSLContext - */ - public SSLContext createSSLContext(Map configuration) { - final SSLMode sslMode = ClientConfigProperties.SSL_MODE.getOrDefault(configuration); - final String trustStorePath = (String) configuration.get(ClientConfigProperties.SSL_TRUST_STORE.getKey()); - final String caCertificate = (String) configuration.get(ClientConfigProperties.CA_CERTIFICATE.getKey()); - final String sslCertificate = (String) configuration.get(ClientConfigProperties.SSL_CERTIFICATE.getKey()); - final String sslKey = (String) configuration.get(ClientConfigProperties.SSL_KEY.getKey()); - - SslContextProvider.Builder builder = sslContextProvider.builder(); - - // The client certificate/key (mTLS) are independent of how the server certificate is verified, - // so they are applied whenever configured, regardless of the SSL mode. - if (sslCertificate != null && !sslCertificate.isEmpty()) { - builder.clientCertificate(sslCertificate, sslKey); - } - - if (sslMode == SSLMode.TRUST) { - // TRUST accepts any server certificate and skips the hostname check (the latter is applied - // where the connection socket factory is created). A configured trust store or CA - // certificate has no effect in this mode and is ignored with a warning. - if (trustStorePath != null || caCertificate != null) { - LOG.warn("SSL mode '{}' trusts any server certificate; the configured {} is ignored.", - SSLMode.TRUST, trustStorePath != null ? "trust store" : "CA certificate"); - } - builder.trustAllCertificates(); - } else if (trustStorePath != null) { - // VERIFY_CA / STRICT: validate against the trust store. A trust store and a CA certificate - // cannot both take effect, so the CA certificate is ignored with a warning. - if (caCertificate != null) { - LOG.warn("Both a trust store and a CA certificate are configured; using the trust store and" - + " ignoring the CA certificate. Import the CA certificate into the trust store instead."); - } - builder.trustStore(trustStorePath, - (String) configuration.get(ClientConfigProperties.SSL_KEY_STORE_PASSWORD.getKey()), - (String) configuration.get(ClientConfigProperties.SSL_KEYSTORE_TYPE.getKey())); - } else if (caCertificate != null) { - // VERIFY_CA / STRICT: validate against the CA certificate. - builder.rootCertificate(caCertificate); - } - // else VERIFY_CA / STRICT with no trust material: the JVM default trust store is used. - - return builder.build(); - } - - private static final long CONNECTION_INACTIVITY_CHECK = 5000L; - - private ConnectionConfig createConnectionConfig(Map configuration) { - ConnectionConfig.Builder connConfig = ConnectionConfig.custom(); - - ClientConfigProperties.CONNECTION_TTL.applyIfSet(configuration, (t) -> connConfig.setTimeToLive(t, TimeUnit.MILLISECONDS)); - ClientConfigProperties.CONNECTION_TIMEOUT.applyIfSet(configuration, (t) -> connConfig.setConnectTimeout(t, TimeUnit.MILLISECONDS)); - connConfig.setValidateAfterInactivity(CONNECTION_INACTIVITY_CHECK, TimeUnit.MILLISECONDS); // non-configurable for now - - return connConfig.build(); - } - - private HttpClientConnectionManager basicConnectionManager(LayeredConnectionSocketFactory sslConnectionSocketFactory, SocketConfig socketConfig, Map configuration) { - Lookup tlsSocketStrategyLookup = RegistryBuilder.create() - .register(URIScheme.HTTPS.id, (socket, target, port, attachment, context) -> - (SSLSocket) sslConnectionSocketFactory.createLayeredSocket(socket, target, port, context)) - .build(); - - BasicHttpClientConnectionManager connManager = BasicHttpClientConnectionManager.create( - null, null, tlsSocketStrategyLookup, null); - connManager.setConnectionConfig(createConnectionConfig(configuration)); - connManager.setSocketConfig(socketConfig); - - return connManager; - } - - private HttpClientConnectionManager poolConnectionManager(LayeredConnectionSocketFactory sslConnectionSocketFactory, SocketConfig socketConfig, Map configuration) { - PoolingHttpClientConnectionManagerBuilder connMgrBuilder = PoolingHttpClientConnectionManagerBuilder.create() - .setPoolConcurrencyPolicy(PoolConcurrencyPolicy.LAX); - - ConnectionReuseStrategy connectionReuseStrategy = ClientConfigProperties.CONNECTION_REUSE_STRATEGY.getOrDefault(configuration); - switch (connectionReuseStrategy) { - case LIFO: - connMgrBuilder.setConnPoolPolicy(PoolReusePolicy.LIFO); - break; - case FIFO: - connMgrBuilder.setConnPoolPolicy(PoolReusePolicy.FIFO); - break; - default: - throw new ClientMisconfigurationException("Unknown connection reuse strategy: " + connectionReuseStrategy); - } - LOG.debug("Connection reuse strategy: {}", connectionReuseStrategy); - - connMgrBuilder.setDefaultConnectionConfig(createConnectionConfig(configuration)); - connMgrBuilder.setMaxConnTotal(Integer.MAX_VALUE); // as we do not know how many routes we will have - ClientConfigProperties.HTTP_MAX_OPEN_CONNECTIONS.applyIfSet(configuration, connMgrBuilder::setMaxConnPerRoute); - - int networkBufferSize = ClientConfigProperties.CLIENT_NETWORK_BUFFER_SIZE.getOrDefault(configuration); - MeteredManagedHttpClientConnectionFactory connectionFactory = new MeteredManagedHttpClientConnectionFactory( - Http1Config.custom() - .setBufferSize(networkBufferSize) - .build(), - CharCodingConfig.DEFAULT, - DefaultHttpResponseParserFactory.INSTANCE); - - connMgrBuilder.setConnectionFactory(connectionFactory); - connMgrBuilder.setSSLSocketFactory(sslConnectionSocketFactory); - connMgrBuilder.setDefaultSocketConfig(socketConfig); - PoolingHttpClientConnectionManager phccm = connMgrBuilder.build(); - poolControl = phccm; - if (metricsRegistry != null) { - try { - String mGroupName = ClientConfigProperties.METRICS_GROUP_NAME.getOrDefault(configuration); - Class micrometerLoader = getClass().getClassLoader().loadClass("com.clickhouse.client.api.metrics.MicrometerLoader"); - Method applyMethod = micrometerLoader.getDeclaredMethod("applyPoolingMetricsBinder", Object.class, String.class, PoolingHttpClientConnectionManager.class); - applyMethod.invoke(micrometerLoader, metricsRegistry, mGroupName, phccm); - - applyMethod = micrometerLoader.getDeclaredMethod("applyConnectionMetricsBinder", Object.class, String.class, MeteredManagedHttpClientConnectionFactory.class); - applyMethod.invoke(micrometerLoader, metricsRegistry, mGroupName, connectionFactory); - } catch (Exception e) { - LOG.error("Failed to register metrics", e); - } - } - return phccm; - } - - public CloseableHttpClient createHttpClient(boolean initSslContext, Map configuration) { - // Top Level builders - HttpClientBuilder clientBuilder = HttpClientBuilder.create(); - // An application-supplied SSLContext is used as is; otherwise one is built from the configured - // trust/key material. Server hostname verification below still applies via the SSL mode. - SSLContext sslContext = null; - if (initSslContext) { - Object customSSLContext = configuration.get(ClientConfigProperties.SSL_CONTEXT.getKey()); - if (customSSLContext == null) { - sslContext = createSSLContext(configuration); - } else if (customSSLContext instanceof SSLContext) { - sslContext = (SSLContext) customSSLContext; - } else { - throw new ClientMisconfigurationException("'" + ClientConfigProperties.SSL_CONTEXT.getKey() - + "' must be a javax.net.ssl.SSLContext instance but was " - + customSSLContext.getClass().getName() + "; supply it via Client.Builder.setSSLContext(...)"); - } - } - LayeredConnectionSocketFactory sslConnectionSocketFactory; - if (sslContext != null) { - String socketSNI = (String)configuration.get(ClientConfigProperties.SSL_SOCKET_SNI.getKey()); - SSLMode sslMode = ClientConfigProperties.SSL_MODE.getOrDefault(configuration); - // Trust and VerifyCa skip hostname verification. The same applies when a custom SNI is - // set because the connection hostname will not match the certificate. - boolean trustAllHostnames = sslMode == SSLMode.TRUST || sslMode == SSLMode.VERIFY_CA; - boolean hasSNI = socketSNI != null && !socketSNI.trim().isEmpty(); - // ssl_cipher_suites is parsed and sanitized in ClientConfigProperties (blank tokens dropped, - // names trimmed); an unset or empty list means "no restriction" so the transport defaults apply. - List cipherSuites = ClientConfigProperties.SSL_CIPHER_SUITES.getOrDefault(configuration); - String[] enabledCipherSuites = cipherSuites == null || cipherSuites.isEmpty() ? null - : cipherSuites.toArray(new String[0]); - if (hasSNI || trustAllHostnames || enabledCipherSuites != null) { - // Skip hostname verification only for trust-all modes or when a custom SNI is used (the - // connection hostname would not match the certificate); otherwise a null verifier makes the - // factory fall back to the JDK/HttpClient default verifier, keeping STRICT verification. - // java:S5527 - the permissive verifier is applied only for SSLMode.TRUST/VERIFY_CA (where the - // user has explicitly opted out of hostname verification) or a custom SNI; STRICT keeps the - // default verifying behaviour, so secure-by-default hostname verification is preserved. - @SuppressWarnings("java:S5527") - HostnameVerifier hostnameVerifier = trustAllHostnames || hasSNI ? (hostname, session) -> true : null; - sslConnectionSocketFactory = new CustomSSLConnectionFactory(socketSNI, sslContext, hostnameVerifier, - enabledCipherSuites); - } else { - sslConnectionSocketFactory = new SSLConnectionSocketFactory(sslContext); - } - } else { - sslConnectionSocketFactory = new DummySSLConnectionSocketFactory(); - } - // Socket configuration - SocketConfig.Builder soCfgBuilder = SocketConfig.custom(); - ClientConfigProperties.SOCKET_OPERATION_TIMEOUT.applyIfSet(configuration, - (t) -> soCfgBuilder.setSoTimeout(t, TimeUnit.MILLISECONDS)); - - ClientConfigProperties.SOCKET_RCVBUF_OPT.applyIfSet(configuration, - soCfgBuilder::setRcvBufSize); - - ClientConfigProperties.SOCKET_SNDBUF_OPT.applyIfSet(configuration, - soCfgBuilder::setSndBufSize); - - ClientConfigProperties.SOCKET_LINGER_OPT.applyIfSet(configuration, - (v) -> soCfgBuilder.setSoLinger(v, TimeUnit.SECONDS)); - - ClientConfigProperties.SOCKET_TCP_NO_DELAY_OPT.applyIfSet(configuration, - soCfgBuilder::setTcpNoDelay); - // Proxy - String proxyHost = (String) configuration.get(ClientConfigProperties.PROXY_HOST.getKey()); - Integer proxyPort = (Integer) configuration.get(ClientConfigProperties.PROXY_PORT.getKey()); - HttpHost proxy = null; - if (proxyHost != null && proxyPort != null) { - proxy = new HttpHost(proxyHost, proxyPort); - } - - String proxyTypeVal = (String) configuration.get(ClientConfigProperties.PROXY_TYPE.getKey()); - ProxyType proxyType = proxyTypeVal == null ? null : ProxyType.valueOf(proxyTypeVal); - if (proxyType == ProxyType.HTTP) { - clientBuilder.setProxy(proxy); - String proxyUser = (String) configuration.get(ClientConfigProperties.PROXY_USER.getKey()); - String proxyPassword = (String) configuration.get(ClientConfigProperties.PROXY_PASSWORD.getKey()); - if (proxyUser != null && proxyPassword != null) { - proxyAuthHeaderValue = "Basic " + Base64.getEncoder().encodeToString((proxyUser + ":" + proxyPassword).getBytes(StandardCharsets.UTF_8)); - } - - } else if (proxyType == ProxyType.SOCKS) { - soCfgBuilder.setSocksProxyAddress(new InetSocketAddress(proxyHost, proxyPort)); - } - - boolean disableCookies = !((Boolean)ClientConfigProperties.HTTP_SAVE_COOKIES.getOrDefault(configuration)); - if (disableCookies) { - clientBuilder.disableCookieManagement(); - } - SocketConfig socketConfig = soCfgBuilder.build(); - - // Connection manager - if (ClientConfigProperties.CONNECTION_POOL_ENABLED.getOrDefault(configuration)) { - clientBuilder.setConnectionManager(poolConnectionManager(sslConnectionSocketFactory, socketConfig, configuration)); - } else { - clientBuilder.setConnectionManager(basicConnectionManager(sslConnectionSocketFactory, socketConfig, configuration)); - } - Long keepAliveTimeout = ClientConfigProperties.HTTP_KEEP_ALIVE_TIMEOUT.getOrDefault(configuration); - if (keepAliveTimeout != null && keepAliveTimeout > 0) { - clientBuilder.setKeepAliveStrategy((response, context) -> TimeValue.ofMilliseconds(keepAliveTimeout)); - } - - clientBuilder.disableContentCompression(); // will handle ourselves - - return clientBuilder.build(); - } - - private static final String ERROR_CODE_PREFIX_PATTERN = "%d. DB::Exception:"; - - /** - * Reads status line and if error tries to parse response body to get server error message. - * - * @param httpResponse - HTTP response - * @return exception object with server code - */ - public Exception readError(HttpPost req, ClassicHttpResponse httpResponse) { - final String queryId = getQueryId(httpResponse, req); - int serverCode = getHeaderInt(httpResponse.getFirstHeader(ClickHouseHttpProto.HEADER_EXCEPTION_CODE), 0); - try { - return serverCode > 0 ? readClickHouseError(httpResponse.getEntity(), serverCode, queryId, httpResponse.getCode()) : - readNotClickHouseError(httpResponse.getEntity(), queryId, httpResponse.getCode()); - } catch (Exception e) { - LOG.error("Failed to read error message", e); - String msg = String.format(ERROR_CODE_PREFIX_PATTERN, serverCode) + " (transport error: " + httpResponse.getCode() + ")"; - return new ServerException(serverCode, msg + " (queryId= " + queryId + ")", httpResponse.getCode(), queryId); - } - } - - private ServerException readNotClickHouseError(HttpEntity httpEntity, String queryId, int httpCode) { - - byte[] buffer = new byte[ERROR_BODY_BUFFER_SIZE]; - - String msg = null; - InputStream body = null; - int offset = 0; - for (int i = 0; i < 2; i++) { - try { - if (body == null) { - body = httpEntity.getContent(); - } - int msgLen = body.read(buffer, offset, buffer.length - offset); - if (msgLen > 0) { - msg = new String(buffer, 0, msgLen, StandardCharsets.UTF_8).trim(); - if (msg.isEmpty()) { - msg = ""; - } - } - break; - } catch (ClientException e) { - // Invalid LZ4 Magic - if (body instanceof CompressedBlockInputStream) { - CompressedBlockInputStream stream = (CompressedBlockInputStream) body; - body = stream.getInputStream(); - byte[] lzHeader = stream.getHeaderBuffer(); // Here is read part of original body - offset = Math.min(lzHeader.length, buffer.length); - System.arraycopy(lzHeader, 0, buffer, 0, offset); - continue; - } - throw e; - } catch (Exception e) { - LOG.warn("Failed to read error message (queryId = " + queryId + ")", e); - break; - } - } - - String errormsg = msg == null ? "unknown server error" : msg; - return new ServerException(ServerException.CODE_UNKNOWN, errormsg + " (transport error: " + httpCode +")", httpCode, queryId); - } - - private static ServerException readClickHouseError(HttpEntity httpEntity, int serverCode, String queryId, int httpCode) throws Exception { - InputStream body = httpEntity.getContent(); - byte[] buffer = new byte[ERROR_BODY_BUFFER_SIZE]; - byte[] lookUpStr = String.format(ERROR_CODE_PREFIX_PATTERN, serverCode).getBytes(StandardCharsets.UTF_8); - StringBuilder msgBuilder = new StringBuilder(); - boolean found = false; - while (true) { - int rBytes = -1; - try { - rBytes = body.read(buffer); - } catch (ClientException e) { - // Invalid LZ4 Magic - if (body instanceof CompressedBlockInputStream) { - CompressedBlockInputStream stream = (CompressedBlockInputStream) body; - body = stream.getInputStream(); - byte[] headerBuffer = stream.getHeaderBuffer(); - System.arraycopy(headerBuffer, 0, buffer, 0, headerBuffer.length); - rBytes = headerBuffer.length; - } - } - if (rBytes == -1) { - break; - } - - for (int i = 0; i < rBytes; i++) { - if (buffer[i] == lookUpStr[0]) { - found = true; - for (int j = 1; j < Math.min(rBytes - i, lookUpStr.length); j++) { - if (buffer[i + j] != lookUpStr[j]) { - found = false; - break; - } - } - if (found) { - msgBuilder.append(new String(buffer, i, rBytes - i, StandardCharsets.UTF_8)); - break; - } - } - } - - if (found) { - break; - } - } - - while (true) { - int rBytes = body.read(buffer); - if (rBytes == -1) { - break; - } - msgBuilder.append(new String(buffer, 0, rBytes, StandardCharsets.UTF_8)); - } - - String msg = msgBuilder.toString().replaceAll("\\s+", " ").replaceAll("\\\\n", " ") - .replaceAll("\\\\/", "/"); - if (msg.trim().isEmpty()) { - msg = String.format(ERROR_CODE_PREFIX_PATTERN, serverCode) + " (transport error: " + httpCode + ")"; - } - return new ServerException(serverCode, "Code: " + msg + " (queryId= " + queryId + ")", httpCode, queryId); - } - - private static final long POOL_VENT_TIMEOUT = 10000L; - private final AtomicLong timeToPoolVent = new AtomicLong(0); - - private void doPoolVent() { - if (poolControl != null && timeToPoolVent.get() < System.currentTimeMillis()) { - timeToPoolVent.set(System.currentTimeMillis() + POOL_VENT_TIMEOUT); - poolControl.closeExpired(); - } - } - - private HttpContext createRequestHttpContext(Map requestConfig) { - HttpClientContext context = HttpClientContext.create(); - Number responseTimeout = ClientConfigProperties.SOCKET_OPERATION_TIMEOUT.getOrDefault(requestConfig); - Number connectionReqTimeout = ClientConfigProperties.CONNECTION_REQUEST_TIMEOUT.getOrDefault(requestConfig); - RequestConfig reqHttpConf = RequestConfig.custom() - .setResponseTimeout(responseTimeout.longValue(), TimeUnit.MILLISECONDS) - .setConnectionRequestTimeout(connectionReqTimeout.longValue(), TimeUnit.MILLISECONDS) - .build(); - context.setRequestConfig(reqHttpConf); - - return context; - } - - private URI createRequestURI(Endpoint server, Map requestConfig, boolean isMultipartRequest) { - URI uri; - try { - URIBuilder uriBuilder = new URIBuilder(server.getURI()); - addRequestParams(requestConfig, uriBuilder::addParameter); - - if (!isMultipartRequest) { - addStatementParams(requestConfig, uriBuilder::addParameter); - } else { - uriBuilder.removeParameter(ClickHouseHttpProto.QPARAM_DECOMPRESS); // multipart request doesn't support compression yet - } - - uri = uriBuilder.optimize().build(); - } catch (URISyntaxException e) { - throw new RuntimeException(e); - } - return uri; - } - - private HttpPost createPostRequest(URI uri, Map requestConfig) { - HttpPost req = new HttpPost(uri); -// req.setVersion(new ProtocolVersion("HTTP", 1, 0)); // to disable chunk transfer encoding - addHeaders(req, requestConfig); - return req; - } - - private static final class TransportRequestImpl implements TransportRequest { - private final HttpPost delegate; - private final Map config; - private final AtomicBoolean cancelled = new AtomicBoolean(false); - - TransportRequestImpl(HttpPost delegate, Map config) { - this.delegate = delegate; - this.config = config; - } - - @Override - public boolean cancel() { - cancelled.set(true); - if (delegate.isCancelled()) { - return true; - } - return delegate.cancel(); - } - - @Override - public boolean isCancelled() { - return cancelled.get(); - } - - @Override - public Map getConfig() { - return config; - } - - @Override - @SuppressWarnings("unchecked") - public T getDelegate() { - return (T) delegate; - } - } - - public TransportRequest createRequest(Endpoint server, Map requestConfig, - String body) { - boolean useMultipart = ClientConfigProperties.HTTP_SEND_PARAMS_IN_BODY.getOrDefault(requestConfig) && - requestConfig.containsKey(HttpAPIClientHelper.KEY_STATEMENT_PARAMS); - - // create configuration dependent objects - final URI uri = createRequestURI(server, requestConfig, useMultipart); - final HttpPost req = createPostRequest(uri, requestConfig); - - final HttpEntity httpEntity; - if (useMultipart) { - // a multipart body is always sent as-is, so the request must not declare a content encoding - the - // server would fail to decompress the plain body. Removed after addHeaders() to also drop an - // encoding set by the application with `http_header_*`. - req.removeHeaders(HttpHeaders.CONTENT_ENCODING); - - MultipartEntityBuilder multipartEntityBuilder = MultipartEntityBuilder.create(); - addStatementParams(requestConfig, multipartEntityBuilder::addTextBody); - multipartEntityBuilder.addTextBody(ClickHouseHttpProto.QPARAM_QUERY_STMT, body); - - httpEntity = multipartEntityBuilder.build(); - req.setHeader(HttpHeaders.CONTENT_TYPE, httpEntity.getContentType()); // set proper content type with generated boundary value - } else { - try { - final String contentEncoding = req.containsHeader(HttpHeaders.CONTENT_ENCODING) ? req.getHeader(HttpHeaders.CONTENT_ENCODING).getValue() : null; - httpEntity = new ByteArrayEntity(body.getBytes(StandardCharsets.UTF_8.name()), CONTENT_TYPE, contentEncoding); - } catch (UnsupportedEncodingException | ProtocolException e) { - throw new ClientException("failed to create request body entity", e); - } - } - // adjust configuration - if (useMultipart) { - req.setEntity(httpEntity); // multipart doesn't support compression right now - } else { - req.setEntity(wrapRequestEntity(httpEntity, requestConfig)); - } - return new TransportRequestImpl(req, requestConfig); - } - - - private static final class TransportResponseImpl implements TransportResponse { - - private final ClassicHttpResponse delegate; - - TransportResponseImpl(ClassicHttpResponse delegate) { - this.delegate = delegate; - } - - @Override - public ClickHouseFormat getDataFormat() { - Header formatHeader = delegate.getFirstHeader(ClickHouseHttpProto.HEADER_FORMAT); - return formatHeader == null ? null : ClickHouseFormat.valueOf(formatHeader.getValue()); - } - - @Override - public String getSummaryJson() { - return HttpAPIClientHelper.getHeaderVal(delegate - .getFirstHeader(ClickHouseHttpProto.HEADER_SRV_SUMMARY), "{}"); - } - - @Override - public String getQueryId() { - return HttpAPIClientHelper.getHeaderVal(delegate - .getFirstHeader(ClickHouseHttpProto.HEADER_QUERY_ID), null); - } - - @Override - @SuppressWarnings("unchecked") - public T getDelegate() { - return (T) delegate; - } - - @Override - public Map getHeaders() { - return HttpAPIClientHelper.collectResponseHeaders(delegate); - } - - @Override - public void close() throws IOException { - delegate.close(); - } - - @Override - public InputStream createDataInputStream() { - try { - return delegate.getEntity().getContent(); - } catch (Exception e) { - throw new ClientException("Failed to construct input stream", e); - } - } - } - - /** - * Executes a single transport request and records it as a child span of the given operation - * span, so a retried operation reports one request span per attempt. The request itself is - * executed by {@link #doExecuteRequest(TransportRequest, Span)}, which stays the single place - * where a request is actually executed and which records the HTTP status of the response. - * - *

- * When no recorder is registered the request is executed by - * {@link #executeRequest(TransportRequest)}, so a client that does not observe spans keeps - * exactly the behaviour it had before and pays nothing for the span path. - * - * @param transportRequest - request to execute - * @param operationSpan - span of the operation this request is made for - * @return transport response - * @throws Exception when the request could not be completed - */ - public TransportResponse executeRequest(TransportRequest transportRequest, Span operationSpan) throws Exception { - if (spanRecorder == DefaultSpanRecorder.NOOP) { - return executeRequest(transportRequest); - } - - final Span requestSpan = startRequestSpan(operationSpan, transportRequest.getDelegate()); - try { - return doExecuteRequest(transportRequest, requestSpan); - } catch (Exception e) { - spanRecorder.recordRequestFailure(requestSpan, e); - throw e; - } finally { - requestSpan.end(); - } - } - - private Span startRequestSpan(Span operationSpan, HttpPost req) { - final URIAuthority authority = req.getAuthority(); - Span span = spanRecorder.startRequestSpan(operationSpan, - authority == null ? null : authority.getHostName(), - authority == null ? -1 : authority.getPort()); - return span == null ? DefaultSpanRecorder.NOOP_SPAN : span; - } - - public TransportResponse executeRequest(TransportRequest transportRequest) throws Exception { - return doExecuteRequest(transportRequest, DefaultSpanRecorder.NOOP_SPAN); - } - - /** - * Executes a single transport request and records the HTTP status on the given request span as - * soon as a response is received - so the status is reported for every response, also for the - * ones this method maps onto an exception that does not carry it (for example {@code 502} and - * {@code 503} onto {@link ConnectException}). - * - * @param transportRequest - request to execute - * @param requestSpan - span of this request; {@link DefaultSpanRecorder#NOOP_SPAN} when the - * request is not recorded - * @return transport response - * @throws Exception when the request could not be completed - */ - private TransportResponse doExecuteRequest(TransportRequest transportRequest, Span requestSpan) throws Exception { - - final Map requestConfig = transportRequest.getConfig(); - final HttpPost req = transportRequest.getDelegate(); - - doPoolVent(); - - ClassicHttpResponse httpResponse = null; - boolean closeResponse = true; - HttpContext context = createRequestHttpContext(requestConfig); - try { - httpResponse = httpClient.executeOpen(null, req, context); - if (requestSpan != DefaultSpanRecorder.NOOP_SPAN) { - // nothing to report when this request is not recorded - and a recorder is never handed a - // span it did not create - spanRecorder.recordHttpStatus(requestSpan, httpResponse.getCode()); - } - - httpResponse.setEntity(wrapResponseEntity(httpResponse.getEntity(), - httpResponse.getCode(), - requestConfig)); - - if (httpResponse.containsHeader(ClickHouseHttpProto.HEADER_EXCEPTION_CODE)) { - throw readError(req, httpResponse); - } - - int statusCode = httpResponse.getCode(); - switch (statusCode) { - case HttpStatus.SC_OK: - closeResponse = false; - return new TransportResponseImpl(httpResponse); - case HttpStatus.SC_PROXY_AUTHENTICATION_REQUIRED: - throw new ClientMisconfigurationException("Proxy authentication required. Please check your proxy settings."); - case HttpStatus.SC_BAD_GATEWAY: - throw new ConnectException("Server returned '502 Bad gateway'. Check network and proxy settings."); - case HttpStatus.SC_SERVICE_UNAVAILABLE: - throw new ConnectException("Server returned '503 Service Unavailable'. Check network settings."); - case HttpStatus.SC_BAD_REQUEST: - case HttpStatus.SC_UNAUTHORIZED: - case HttpStatus.SC_FORBIDDEN: - case HttpStatus.SC_SERVER_ERROR: - case HttpStatus.SC_NOT_FOUND: - // ClickHouse usually uses SC_BAD_REQUEST and SC_SERVER_ERROR to return error. - // SC_UNAUTHORIZED, SC_FORBIDDEN is for authentication - // SC_NOT_FOUND can be returned by ClickHouse when path doesn't match database, but also by proxy - // others we cannot handle properly - throw readError(req, httpResponse); - default: - // Unknown status code: log it once (the generic exception below carries no server context). - logServerErrorResponse(req, httpResponse); - throw new ClientException("Unexpected result status " + statusCode); - } - } catch (UnknownHostException e) { - LOG.debug("Host '{}' unknown", req.getAuthority()); - throw e; - } catch (ConnectException | NoRouteToHostException e) { - LOG.debug("Failed to connect to '{}': {}", req.getAuthority(), e.getMessage()); - throw e; - } catch (Exception e) { - LOG.debug("Failed to execute request to '{}': {}", req.getAuthority(), e.getMessage(), e); - if (e instanceof RequestFailedException && req.isCancelled()) { - throw new TransportException("Request was cancelled on client side", e, getQueryId(httpResponse, req)); - } - throw e; - } finally { - if (closeResponse) { - ClientUtils.quietClose(httpResponse, LOG); - } - } - } - - public TransportRequest createRequest(Endpoint server, Map requestConfig, IOCallback writeCallback) { - final URI uri = createRequestURI(server, requestConfig, false); - final HttpPost req = createPostRequest(uri, requestConfig); - try { - String contentEncoding = req.containsHeader(HttpHeaders.CONTENT_ENCODING) ? req.getHeader(HttpHeaders.CONTENT_ENCODING).getValue() : null; - req.setEntity(wrapRequestEntity( - new EntityTemplate(-1, CONTENT_TYPE, contentEncoding, writeCallback), - requestConfig)); - } catch (ProtocolException e) { - throw new ClientException("failed to create request body entity", e); - } - - return new TransportRequestImpl(req, requestConfig); - } - - private String getQueryId(HttpResponse httpResponse, HttpPost httpRequest) { - final Header serverQueryIdHeader = httpResponse == null ? null : httpResponse.getFirstHeader(ClickHouseHttpProto.HEADER_QUERY_ID); - final Header clientQueryIdHeader = httpRequest == null ? null : httpRequest.getFirstHeader(ClickHouseHttpProto.HEADER_QUERY_ID); - final Header queryHeader = Stream.of(serverQueryIdHeader, clientQueryIdHeader).filter(Objects::nonNull).findFirst().orElse(null); - return queryHeader == null ? "" : queryHeader.getValue(); - } - - /** - * Logs a server error response at WARN (status, query id, authority, exception-code header). The body is - * not logged: {@link #readError} consumes it and it may contain SQL/data. - */ - private void logServerErrorResponse(HttpPost req, ClassicHttpResponse httpResponse) { - // Null-safe: the error path must never let its own logger throw and mask the real failure. - if (req == null || httpResponse == null) { - return; - } - final Header exceptionCodeHeader = httpResponse.getFirstHeader(ClickHouseHttpProto.HEADER_EXCEPTION_CODE); - LOG.warn("Server returned error response: status={}, queryId='{}', authority='{}', {}={}", - httpResponse.getCode(), - getQueryId(httpResponse, req), - req.getAuthority(), - ClickHouseHttpProto.HEADER_EXCEPTION_CODE, - exceptionCodeHeader == null ? "" : exceptionCodeHeader.getValue()); - } - - private static final ContentType CONTENT_TYPE = ContentType.create(ContentType.TEXT_PLAIN.getMimeType(), "UTF-8"); - - private void addHeaders(HttpPost req, Map requestConfig) { - setHeader(req, HttpHeaders.CONTENT_TYPE, CONTENT_TYPE.getMimeType()); - String socketSNI = (String) requestConfig.get(ClientConfigProperties.SSL_SOCKET_SNI.getKey()); - if (ClientUtils.isNotBlank(socketSNI) && !req.containsHeader(HttpHeaders.HOST)) { - int port = -1; - try { - if (req.getUri() != null) { - port = req.getUri().getPort(); - } - } catch (Exception ignore) { - // ignore - } - String hostHeader = (port > 0 && port != 80 && port != 443) ? socketSNI + ":" + port : socketSNI; - setHeader(req, HttpHeaders.HOST, hostHeader); - } - if (requestConfig.containsKey(ClientConfigProperties.INPUT_OUTPUT_FORMAT.getKey())) { - Object formatObj = requestConfig.get(ClientConfigProperties.INPUT_OUTPUT_FORMAT.getKey()); - if (formatObj != null) { - String formatStr = formatObj instanceof String ? formatObj.toString() : ((ClickHouseFormat)formatObj).name(); - if (ClientUtils.isNotBlank(formatStr)) { - setHeader( - req, - ClickHouseHttpProto.HEADER_FORMAT, - formatStr); - } - } - } - if (requestConfig.containsKey(ClientConfigProperties.QUERY_ID.getKey())) { - setHeader( - req, - ClickHouseHttpProto.HEADER_QUERY_ID, - (String) requestConfig.get(ClientConfigProperties.QUERY_ID.getKey())); - } - setHeader( - req, - ClickHouseHttpProto.HEADER_DATABASE, - ClientConfigProperties.DATABASE.getOrDefault(requestConfig)); - - if (ClientConfigProperties.SSL_AUTH.getOrDefault(requestConfig).booleanValue()) { - setHeader( - req, - ClickHouseHttpProto.HEADER_DB_USER, - ClientConfigProperties.USER.getOrDefault(requestConfig)); - setHeader( - req, - ClickHouseHttpProto.HEADER_SSL_CERT_AUTH, - "on"); - } else if (ClientConfigProperties.HTTP_USE_BASIC_AUTH.getOrDefault(requestConfig).booleanValue()) { - String user = ClientConfigProperties.USER.getOrDefault(requestConfig); - String password = ClientConfigProperties.PASSWORD.getOrDefault(requestConfig); - // Use as-is, no encoding allowed - req.addHeader( - HttpHeaders.AUTHORIZATION, - "Basic " + Base64.getEncoder().encodeToString( - (user + ":" + password).getBytes(StandardCharsets.UTF_8))); - } else { - setHeader( - req, - ClickHouseHttpProto.HEADER_DB_USER, - ClientConfigProperties.USER.getOrDefault(requestConfig)); - setHeader( - req, - ClickHouseHttpProto.HEADER_DB_PASSWORD, - ClientConfigProperties.PASSWORD.getOrDefault(requestConfig)); - } - if (proxyAuthHeaderValue != null) { - req.addHeader( - HttpHeaders.PROXY_AUTHORIZATION, - proxyAuthHeaderValue); - } - - boolean clientCompression = ClientConfigProperties.COMPRESS_CLIENT_REQUEST.getOrDefault(requestConfig); - boolean serverCompression = ClientConfigProperties.COMPRESS_SERVER_RESPONSE.getOrDefault(requestConfig); - boolean useHttpCompression = ClientConfigProperties.USE_HTTP_COMPRESSION.getOrDefault(requestConfig); - boolean appCompressedData = ClientConfigProperties.APP_COMPRESSED_DATA.getOrDefault(requestConfig); - - if (useHttpCompression) { - if (serverCompression) { - setHeader(req, HttpHeaders.ACCEPT_ENCODING, DEFAULT_HTTP_COMPRESSION_ALGO); - } - - if (clientCompression && !appCompressedData) { - setHeader(req, HttpHeaders.CONTENT_ENCODING, DEFAULT_HTTP_COMPRESSION_ALGO); - } - } - - for (String key : requestConfig.keySet()) { - if (key.startsWith(ClientConfigProperties.HTTP_HEADER_PREFIX)) { - Object val = requestConfig.get(key); - if (val != null) { - setHeader( - req, - key.substring(ClientConfigProperties.HTTP_HEADER_PREFIX.length()), - String.valueOf(val)); - } - } - } - - // Special cases - if (req.containsHeader(HttpHeaders.AUTHORIZATION) - && (req.containsHeader(ClickHouseHttpProto.HEADER_DB_USER) || - req.containsHeader(ClickHouseHttpProto.HEADER_DB_PASSWORD))) - { - // user has set auth header for purpose, lets remove ours - req.removeHeaders(ClickHouseHttpProto.HEADER_DB_USER); - req.removeHeaders(ClickHouseHttpProto.HEADER_DB_PASSWORD); - } - - // -- keep last - correctUserAgentHeader(req, requestConfig); - } - - private void addRequestParams(Map requestConfig, BiConsumer consumer) { - if (requestConfig.containsKey(ClientConfigProperties.QUERY_ID.getKey())) { - consumer.accept(ClickHouseHttpProto.QPARAM_QUERY_ID, requestConfig.get(ClientConfigProperties.QUERY_ID.getKey()).toString()); - } - - boolean clientCompression = ClientConfigProperties.COMPRESS_CLIENT_REQUEST.getOrDefault(requestConfig); - boolean serverCompression = ClientConfigProperties.COMPRESS_SERVER_RESPONSE.getOrDefault(requestConfig); - boolean useHttpCompression = ClientConfigProperties.USE_HTTP_COMPRESSION.getOrDefault(requestConfig); - - if (useHttpCompression) { - // enable_http_compression make server react on http header - // for client side compression Content-Encoding should be set - // for server side compression Accept-Encoding should be set - consumer.accept(ClickHouseHttpProto.QPARAM_ENABLE_HTTP_COMPRESSION, "1"); - } else { - if (serverCompression) { - consumer.accept(ClickHouseHttpProto.QPARAM_COMPRESS, "1"); - } - if (clientCompression) { - consumer.accept(ClickHouseHttpProto.QPARAM_DECOMPRESS, "1"); - } - } - - Collection sessionRoles = ClientConfigProperties.SESSION_DB_ROLES.getOrDefault(requestConfig); - if (!(sessionRoles == null || sessionRoles.isEmpty())) { - sessionRoles.forEach(r -> consumer.accept(ClickHouseHttpProto.QPARAM_ROLE, r)); - } - - for (String key : requestConfig.keySet()) { - if (key.startsWith(ClientConfigProperties.SERVER_SETTING_PREFIX)) { - Object val = requestConfig.get(key); - if (val != null) { - consumer.accept(key.substring(ClientConfigProperties.SERVER_SETTING_PREFIX.length()), String.valueOf(requestConfig.get(key))); - } - } - } - } - - private void addStatementParams(Map requestConfig, BiConsumer consumer) { - if (requestConfig.containsKey(KEY_STATEMENT_PARAMS)) { - Map params = (Map) requestConfig.get(KEY_STATEMENT_PARAMS); - params.forEach((k, v) -> consumer.accept("param_" + k, String.valueOf(v))); - } - } - - private HttpEntity wrapRequestEntity(HttpEntity httpEntity, Map requestConfig) { - - boolean clientCompression = ClientConfigProperties.COMPRESS_CLIENT_REQUEST.getOrDefault(requestConfig); - boolean useHttpCompression = ClientConfigProperties.USE_HTTP_COMPRESSION.getOrDefault(requestConfig); - boolean appCompressedData = ClientConfigProperties.APP_COMPRESSED_DATA.getOrDefault(requestConfig); - - if (httpEntity.getContentEncoding() != null && !appCompressedData) { - // http header is set and data is not compressed - return new CompressedEntity(httpEntity, false, CompressorStreamFactory.getSingleton()); - } else if (clientCompression && !appCompressedData) { - int buffSize = ClientConfigProperties.COMPRESSION_LZ4_UNCOMPRESSED_BUF_SIZE.getOrDefault(requestConfig); - CompressionMethod compressMethod = ClientConfigProperties.COMPRESSION_METHOD.getOrDefault(requestConfig); - return new CompressedBlockEntity(httpEntity, useHttpCompression, false, true, - buffSize, false, lz4Factory, compressMethod); - } else { - return httpEntity; - } - } - - private HttpEntity wrapResponseEntity(HttpEntity httpEntity, int httpStatus, Map requestConfig) { - boolean serverCompression = ClientConfigProperties.COMPRESS_SERVER_RESPONSE.getOrDefault(requestConfig); - boolean useHttpCompression = ClientConfigProperties.USE_HTTP_COMPRESSION.getOrDefault(requestConfig); - - if (httpEntity.getContentEncoding() != null) { - // http compressed response - return new CompressedEntity(httpEntity, true, CompressorStreamFactory.getSingleton()); - } - - // data compression - if (serverCompression && !(httpStatus == HttpStatus.SC_FORBIDDEN || httpStatus == HttpStatus.SC_UNAUTHORIZED)) { - int buffSize = ClientConfigProperties.COMPRESSION_LZ4_UNCOMPRESSED_BUF_SIZE.getOrDefault(requestConfig); - return new CompressedBlockEntity(httpEntity, useHttpCompression, true, false, buffSize, true, lz4Factory, - ClientConfigProperties.COMPRESSION_METHOD.getDefObjVal()); - } - - return httpEntity; - } - - public static int getHeaderInt(Header header, int defaultValue) { - return getHeaderVal(header, defaultValue, Integer::parseInt); - } - - private static final Set RESPONSE_HEADER_WHITELIST = new HashSet<>(Arrays.asList( - ClickHouseHttpProto.HEADER_QUERY_ID, - ClickHouseHttpProto.HEADER_SRV_SUMMARY, - ClickHouseHttpProto.HEADER_SRV_DISPLAY_NAME, - ClickHouseHttpProto.HEADER_DATABASE, - ClickHouseHttpProto.HEADER_DB_USER, - ClickHouseHttpProto.HEADER_TIMEZONE, - ClickHouseHttpProto.HEADER_FORMAT, - ClickHouseHttpProto.HEADER_PROGRESS - )); - - /** - * Collects whitelisted response headers from an HTTP response into a map. - * - * @param response the HTTP response - * @return unmodifiable map of header name to header value for whitelisted headers present in the response - */ - public static Map collectResponseHeaders(ClassicHttpResponse response) { - Map headers = new HashMap<>(); - for (String name : RESPONSE_HEADER_WHITELIST) { - Header header = response.getFirstHeader(name); - if (header != null) { - headers.put(name, header.getValue()); - } - } - return Collections.unmodifiableMap(headers); - } - - public static String getHeaderVal(Header header, String defaultValue) { - return getHeaderVal(header, defaultValue, Function.identity()); - } - - public static T getHeaderVal(Header header, T defaultValue, Function converter) { - if (header == null) { - return defaultValue; - } - - return converter.apply(header.getValue()); - } - - public boolean shouldRetry(Throwable ex, Map requestSettings) { - List retryCauses = ClientConfigProperties.CLIENT_RETRY_ON_FAILURE.getOrDefault(requestSettings); - - if (retryCauses.contains(ClientFaultCause.None)) { - return false; - } - - if (ex instanceof NoHttpResponseException - || ex.getCause() instanceof NoHttpResponseException) { - return retryCauses.contains(ClientFaultCause.NoHttpResponse); - } - - if (ex instanceof ConnectException - || ex instanceof ConnectTimeoutException - || ex.getCause() instanceof ConnectException - || ex.getCause() instanceof ConnectTimeoutException) { - return retryCauses.contains(ClientFaultCause.ConnectTimeout); - } - - if (ex instanceof ConnectionRequestTimeoutException - || ex.getCause() instanceof ConnectionRequestTimeoutException) { - return retryCauses.contains(ClientFaultCause.ConnectionRequestTimeout); - } - - if (ex instanceof SocketTimeoutException - || ex.getCause() instanceof SocketTimeoutException) { - return retryCauses.contains(ClientFaultCause.SocketTimeout); - } - - // there are some db retryable error codes - if (ex instanceof ServerException || ex.getCause() instanceof ServerException) { - ServerException se = (ServerException) (ex instanceof ServerException ? ex : ex.getCause()); - if (se.getCode() == ServerException.EXECUTION_TIMEOUT) { - return retryCauses.contains(ClientFaultCause.ServerTimeoutExceeded); - } - return se.isRetryable() && retryCauses.contains(ClientFaultCause.ServerRetryable); - } - - return false; - } - - // This method wraps some client specific exceptions into specific ClientException or just ClientException - // ClientException will be also wrapped - public RuntimeException wrapException(String message, Exception cause, String queryId) { - // Already-classified exceptions (ClientException, ServerException, ConnectionInitiationException, ...) - // are returned as-is so their specific type is preserved instead of being reboxed as a generic - // ClickHouseException. - if (cause instanceof ClickHouseException) { - return (ClickHouseException) cause; - } - - if (cause instanceof SSLException) { - return new TransportException("SSL Problem", cause, queryId); - } - - if (cause instanceof ConnectionRequestTimeoutException || - cause instanceof NoHttpResponseException || - cause instanceof ConnectTimeoutException || - cause instanceof ConnectException || - cause instanceof UnknownHostException || - cause instanceof NoRouteToHostException) { - ConnectionInitiationException ex = new ConnectionInitiationException(message, cause); - ex.setQueryId(queryId); - return ex; - } - - if (cause instanceof SocketTimeoutException || cause instanceof IOException) { - DataTransferException ex = new DataTransferException(message, cause); - ex.setQueryId(queryId); - return ex; - } - // if we can not identify the exception explicitly we catch as our base exception ClickHouseException - return new ClickHouseException(message, cause, queryId); - } - - private void correctUserAgentHeader(HttpRequest request, Map requestConfig) { - //TODO: implement cache for user-agent - Header userAgentHeader = request.getLastHeader(HttpHeaders.USER_AGENT); - request.removeHeaders(HttpHeaders.USER_AGENT); - - String clientName = ClientConfigProperties.CLIENT_NAME.getOrDefault(requestConfig); - String userAgentValue = defaultUserAgent; - if (userAgentHeader == null && clientName != null && !clientName.isEmpty()) { - userAgentValue = clientName + " " + defaultUserAgent; - } else if (userAgentHeader != null) { - userAgentValue = userAgentHeader.getValue() + " " + defaultUserAgent; - } - request.setHeader(HttpHeaders.USER_AGENT, userAgentValue); - } - - private String buildDefaultUserAgent() { - StringBuilder userAgent = new StringBuilder(); - userAgent.append(Client.CLIENT_USER_AGENT); - - String clientVersion = Client.clientVersion; - - userAgent.append(clientVersion); - - userAgent.append(" ("); - userAgent.append(System.getProperty("os.name")); - userAgent.append("; "); - userAgent.append("jvm:").append(System.getProperty("java.version")); - userAgent.append("; "); - - userAgent.setLength(userAgent.length() - 2); - userAgent.append(')'); - - try { - String httpClientVersion = this.httpClient.getClass().getPackage().getImplementationVersion(); - if (Objects.equals(this.httpClient.getClass().getPackage().getImplementationTitle(), this.getClass().getPackage().getImplementationTitle())) { - // shaded jar - all packages have same implementation title - httpClientVersion = "unknown"; - try (InputStream in = Thread.currentThread().getContextClassLoader().getResourceAsStream("client-v2-version.properties")) { - Properties p = new Properties(); - p.load(in); - - String tmp = p.getProperty("apache.http.client.version"); - if (tmp != null && !tmp.isEmpty() && !tmp.equals("${apache.httpclient.version}")) { - httpClientVersion = tmp; - } - } catch (Exception e) { - LOG.debug("Failed to read HTTP client version from client-v2-version.properties", e); - } - } - userAgent.append(" ") - .append("Apache-HttpClient") - .append('/') - .append(httpClientVersion); - } catch (Exception e) { - LOG.warn("failed to construct http client version string", e); - } - return userAgent.toString(); - } - - public void close() { - httpClient.close(CloseMode.IMMEDIATE); - } - - private static void setHeader(HttpRequest req, String headerName, - String value) - { - if (value == null) { - return; - } - - if (value.trim().isEmpty()) { - return; - } - if (PATTERN_HEADER_VALUE_ASCII.matcher(value).matches()) { - req.setHeader(headerName, value); - } else { - try { - req.setHeader( - headerName + "*", - "UTF-8''" + URLEncoder.encode(value, StandardCharsets.UTF_8.name())); - } catch (UnsupportedEncodingException e) { - throw new ClientException("Failed to convert string to UTF8" , e); - } - } - } - - /** - * This factory is used only when no ssl connections are required (no https endpoints). - * Internally http client would create factory and spend time if no supplied. - */ - private static class DummySSLConnectionSocketFactory implements LayeredConnectionSocketFactory { - @Override - public Socket createLayeredSocket(Socket socket, String target, int port, HttpContext context) throws IOException { - return null; - } - - @Override - public Socket createSocket(HttpContext context) throws IOException { - return null; - } - - @Override - public Socket connectSocket(TimeValue connectTimeout, Socket socket, HttpHost host, InetSocketAddress remoteAddress, InetSocketAddress localAddress, HttpContext context) throws IOException { - return null; - } - } - - public static class MeteredManagedHttpClientConnectionFactory extends ManagedHttpClientConnectionFactory { - public MeteredManagedHttpClientConnectionFactory(Http1Config http1Config, CharCodingConfig charCodingConfig, DefaultHttpResponseParserFactory defaultHttpResponseParserFactory) { - super(http1Config, charCodingConfig, defaultHttpResponseParserFactory); - } - - ConcurrentLinkedQueue times = new ConcurrentLinkedQueue<>(); - - - @Override - public ManagedHttpClientConnection createConnection(Socket socket) throws IOException { - long startT = System.currentTimeMillis(); - try { - return super.createConnection(socket); - } finally { - long endT = System.currentTimeMillis(); - times.add(endT - startT); - } - } - - public long getTime() { - int count = times.size(); - long runningAverage = 0; - for (int i = 0; i < count; i++) { - Long t = times.poll(); - if (t != null) { - runningAverage += t; - } - } - - return count > 0 ? runningAverage / count : 0; - } - } - - public static class CustomSSLConnectionFactory extends SSLConnectionSocketFactory { - - private final SNIHostName defaultSNI; - - // Retained for backward compatibility; delegates with no cipher-suite restriction (transport defaults). - public CustomSSLConnectionFactory(String defaultSNI, SSLContext sslContext, HostnameVerifier hostnameVerifier) { - this(defaultSNI, sslContext, hostnameVerifier, null); - } - - public CustomSSLConnectionFactory(String defaultSNI, SSLContext sslContext, HostnameVerifier hostnameVerifier, - String[] supportedCipherSuites) { - super(sslContext, null /* supportedProtocols */, supportedCipherSuites, hostnameVerifier); - this.defaultSNI = defaultSNI == null || defaultSNI.trim().isEmpty() ? null : new SNIHostName(defaultSNI); - } - - @Override - protected void prepareSocket(SSLSocket socket, HttpContext context) throws IOException { - super.prepareSocket(socket, context); - - if (defaultSNI != null) { - SSLParameters sslParams = socket.getSSLParameters(); - sslParams.setServerNames(Collections.singletonList(defaultSNI)); - socket.setSSLParameters(sslParams); - } - } - } -} +package com.clickhouse.client.api.internal; + +import com.clickhouse.client.api.ClickHouseException; +import com.clickhouse.client.api.Client; +import com.clickhouse.client.api.ClientConfigProperties; +import com.clickhouse.client.api.ClientException; +import com.clickhouse.client.api.ClientFaultCause; +import com.clickhouse.client.api.ClientMisconfigurationException; +import com.clickhouse.client.api.CompressionMethod; +import com.clickhouse.client.api.ConnectionInitiationException; +import com.clickhouse.client.api.ConnectionReuseStrategy; +import com.clickhouse.client.api.DataTransferException; +import com.clickhouse.client.api.ServerException; +import com.clickhouse.client.api.TransportException; +import com.clickhouse.client.api.enums.ProxyType; +import com.clickhouse.client.api.enums.SSLMode; +import com.clickhouse.client.api.http.ClickHouseHttpProto; +import com.clickhouse.client.api.observability.DefaultSpanRecorder; +import com.clickhouse.client.api.observability.Span; +import com.clickhouse.client.api.observability.SpanRecorder; +import com.clickhouse.client.api.transport.Endpoint; +import com.clickhouse.client.api.transport.internal.TransportRequest; +import com.clickhouse.client.api.transport.internal.TransportResponse; +import com.clickhouse.data.ClickHouseFormat; +import net.jpountz.lz4.LZ4Factory; +import org.apache.commons.compress.compressors.CompressorStreamFactory; +import org.apache.hc.client5.http.ConnectTimeoutException; +import org.apache.hc.client5.http.classic.methods.HttpPost; +import org.apache.hc.client5.http.config.ConnectionConfig; +import org.apache.hc.client5.http.config.RequestConfig; +import org.apache.hc.client5.http.entity.mime.MultipartEntityBuilder; +import org.apache.hc.client5.http.impl.classic.CloseableHttpClient; +import org.apache.hc.client5.http.impl.classic.HttpClientBuilder; +import org.apache.hc.client5.http.impl.classic.RequestFailedException; +import org.apache.hc.client5.http.impl.io.BasicHttpClientConnectionManager; +import org.apache.hc.client5.http.impl.io.ManagedHttpClientConnectionFactory; +import org.apache.hc.client5.http.impl.io.PoolingHttpClientConnectionManager; +import org.apache.hc.client5.http.impl.io.PoolingHttpClientConnectionManagerBuilder; +import org.apache.hc.client5.http.io.HttpClientConnectionManager; +import org.apache.hc.client5.http.io.ManagedHttpClientConnection; +import org.apache.hc.client5.http.protocol.HttpClientContext; +import org.apache.hc.client5.http.socket.LayeredConnectionSocketFactory; +import org.apache.hc.client5.http.ssl.SSLConnectionSocketFactory; +import org.apache.hc.client5.http.ssl.TlsSocketStrategy; +import org.apache.hc.core5.http.ClassicHttpResponse; +import org.apache.hc.core5.http.ConnectionRequestTimeoutException; +import org.apache.hc.core5.http.ContentType; +import org.apache.hc.core5.http.Header; +import org.apache.hc.core5.http.HttpEntity; +import org.apache.hc.core5.http.HttpHeaders; +import org.apache.hc.core5.http.HttpHost; +import org.apache.hc.core5.http.HttpRequest; +import org.apache.hc.core5.http.HttpResponse; +import org.apache.hc.core5.http.HttpStatus; +import org.apache.hc.core5.http.NoHttpResponseException; +import org.apache.hc.core5.http.ProtocolException; +import org.apache.hc.core5.http.URIScheme; +import org.apache.hc.core5.http.config.CharCodingConfig; +import org.apache.hc.core5.http.config.Http1Config; +import org.apache.hc.core5.http.config.Lookup; +import org.apache.hc.core5.http.config.RegistryBuilder; +import org.apache.hc.core5.http.impl.io.DefaultHttpResponseParserFactory; +import org.apache.hc.core5.http.io.SocketConfig; +import org.apache.hc.core5.http.io.entity.ByteArrayEntity; +import org.apache.hc.core5.http.io.entity.EntityTemplate; +import org.apache.hc.core5.http.protocol.HttpContext; +import org.apache.hc.core5.io.CloseMode; +import org.apache.hc.core5.io.IOCallback; +import org.apache.hc.core5.net.URIAuthority; +import org.apache.hc.core5.net.URIBuilder; +import org.apache.hc.core5.pool.ConnPoolControl; +import org.apache.hc.core5.pool.PoolConcurrencyPolicy; +import org.apache.hc.core5.pool.PoolReusePolicy; +import org.apache.hc.core5.util.TimeValue; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.net.ssl.HostnameVerifier; +import javax.net.ssl.SNIHostName; +import javax.net.ssl.SSLContext; +import javax.net.ssl.SSLException; +import javax.net.ssl.SSLParameters; +import javax.net.ssl.SSLSocket; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.io.UnsupportedEncodingException; +import java.lang.reflect.Method; +import java.net.ConnectException; +import java.net.InetSocketAddress; +import java.net.NoRouteToHostException; +import java.net.Socket; +import java.net.SocketTimeoutException; +import java.net.URI; +import java.net.URISyntaxException; +import java.net.URLEncoder; +import java.net.UnknownHostException; +import java.nio.charset.StandardCharsets; +import java.util.Arrays; +import java.util.Base64; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Properties; +import java.util.Set; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; +import java.util.function.BiConsumer; +import java.util.function.Function; +import java.util.regex.Pattern; +import java.util.stream.Stream; + +public class HttpAPIClientHelper { + + public static final String KEY_STATEMENT_PARAMS = "statement_params"; + + private static final Logger LOG = LoggerFactory.getLogger(HttpAPIClientHelper.class); + + private static final int ERROR_BODY_BUFFER_SIZE = 1024; // Error messages are usually small + + private final String DEFAULT_HTTP_COMPRESSION_ALGO = "lz4"; + + private static final Pattern PATTERN_HEADER_VALUE_ASCII =Pattern.compile("[\\x21-\\x7E]+(?:[ \\t]+[\\x21-\\x7E]+)*"); + + private final CloseableHttpClient httpClient; + + private String proxyAuthHeaderValue; + + private final Set defaultRetryCauses; + + private final String defaultUserAgent; + private final Object metricsRegistry; + + ConnPoolControl poolControl; + + LZ4Factory lz4Factory; + + private final SslContextProvider sslContextProvider = new SslContextProvider(); + + /** + * Recorder a span is started on per transport request. Never {@code null} - + * {@link com.clickhouse.client.api.observability.DefaultSpanRecorder#NOOP} when observability is + * not configured. + */ + private final SpanRecorder spanRecorder; + + public HttpAPIClientHelper(Map configuration, Object metricsRegistry, boolean initSslContext, + LZ4Factory lz4Factory, SpanRecorder spanRecorder) { + this.spanRecorder = Objects.requireNonNull(spanRecorder, + "spanRecorder is required; use DefaultSpanRecorder.NOOP to record nothing"); + this.metricsRegistry = metricsRegistry; + this.httpClient = createHttpClient(initSslContext, configuration); + this.lz4Factory = lz4Factory; + assert this.lz4Factory != null; + + boolean usingClientCompression = ClientConfigProperties.COMPRESS_CLIENT_REQUEST.getOrDefault(configuration); + boolean usingServerCompression = ClientConfigProperties.COMPRESS_SERVER_RESPONSE.getOrDefault(configuration); + boolean useHttpCompression = ClientConfigProperties.USE_HTTP_COMPRESSION.getOrDefault(configuration); + + LOG.debug("client compression: {}, server compression: {}, http compression: {}", usingClientCompression, usingServerCompression, useHttpCompression); + + defaultRetryCauses = new HashSet<>(ClientConfigProperties.CLIENT_RETRY_ON_FAILURE.getOrDefault(configuration)); + if (defaultRetryCauses.contains(ClientFaultCause.None)) { + defaultRetryCauses.removeIf(c -> c != ClientFaultCause.None); + } + + this.defaultUserAgent = buildDefaultUserAgent(); + } + + /** + * Creates an SSL context from the configured trust/key material. + * + * @return SSLContext + */ + public SSLContext createSSLContext(Map configuration) { + final SSLMode sslMode = ClientConfigProperties.SSL_MODE.getOrDefault(configuration); + final String trustStorePath = (String) configuration.get(ClientConfigProperties.SSL_TRUST_STORE.getKey()); + final String caCertificate = (String) configuration.get(ClientConfigProperties.CA_CERTIFICATE.getKey()); + final String sslCertificate = (String) configuration.get(ClientConfigProperties.SSL_CERTIFICATE.getKey()); + final String sslKey = (String) configuration.get(ClientConfigProperties.SSL_KEY.getKey()); + + SslContextProvider.Builder builder = sslContextProvider.builder(); + + // The client certificate/key (mTLS) are independent of how the server certificate is verified, + // so they are applied whenever configured, regardless of the SSL mode. + if (sslCertificate != null && !sslCertificate.isEmpty()) { + builder.clientCertificate(sslCertificate, sslKey); + } + + if (sslMode == SSLMode.TRUST) { + // TRUST accepts any server certificate and skips the hostname check (the latter is applied + // where the connection socket factory is created). A configured trust store or CA + // certificate has no effect in this mode and is ignored with a warning. + if (trustStorePath != null || caCertificate != null) { + LOG.warn("SSL mode '{}' trusts any server certificate; the configured {} is ignored.", + SSLMode.TRUST, trustStorePath != null ? "trust store" : "CA certificate"); + } + builder.trustAllCertificates(); + } else if (trustStorePath != null) { + // VERIFY_CA / STRICT: validate against the trust store. A trust store and a CA certificate + // cannot both take effect, so the CA certificate is ignored with a warning. + if (caCertificate != null) { + LOG.warn("Both a trust store and a CA certificate are configured; using the trust store and" + + " ignoring the CA certificate. Import the CA certificate into the trust store instead."); + } + builder.trustStore(trustStorePath, + (String) configuration.get(ClientConfigProperties.SSL_KEY_STORE_PASSWORD.getKey()), + (String) configuration.get(ClientConfigProperties.SSL_KEYSTORE_TYPE.getKey())); + } else if (caCertificate != null) { + // VERIFY_CA / STRICT: validate against the CA certificate. + builder.rootCertificate(caCertificate); + } + // else VERIFY_CA / STRICT with no trust material: the JVM default trust store is used. + + return builder.build(); + } + + private static final long CONNECTION_INACTIVITY_CHECK = 5000L; + + private ConnectionConfig createConnectionConfig(Map configuration) { + ConnectionConfig.Builder connConfig = ConnectionConfig.custom(); + + ClientConfigProperties.CONNECTION_TTL.applyIfSet(configuration, (t) -> connConfig.setTimeToLive(t, TimeUnit.MILLISECONDS)); + ClientConfigProperties.CONNECTION_TIMEOUT.applyIfSet(configuration, (t) -> connConfig.setConnectTimeout(t, TimeUnit.MILLISECONDS)); + connConfig.setValidateAfterInactivity(CONNECTION_INACTIVITY_CHECK, TimeUnit.MILLISECONDS); // non-configurable for now + + return connConfig.build(); + } + + private HttpClientConnectionManager basicConnectionManager(LayeredConnectionSocketFactory sslConnectionSocketFactory, SocketConfig socketConfig, Map configuration) { + Lookup tlsSocketStrategyLookup = RegistryBuilder.create() + .register(URIScheme.HTTPS.id, (socket, target, port, attachment, context) -> + (SSLSocket) sslConnectionSocketFactory.createLayeredSocket(socket, target, port, context)) + .build(); + + BasicHttpClientConnectionManager connManager = BasicHttpClientConnectionManager.create( + null, null, tlsSocketStrategyLookup, null); + connManager.setConnectionConfig(createConnectionConfig(configuration)); + connManager.setSocketConfig(socketConfig); + + return connManager; + } + + private HttpClientConnectionManager poolConnectionManager(LayeredConnectionSocketFactory sslConnectionSocketFactory, SocketConfig socketConfig, Map configuration) { + PoolingHttpClientConnectionManagerBuilder connMgrBuilder = PoolingHttpClientConnectionManagerBuilder.create() + .setPoolConcurrencyPolicy(PoolConcurrencyPolicy.LAX); + + ConnectionReuseStrategy connectionReuseStrategy = ClientConfigProperties.CONNECTION_REUSE_STRATEGY.getOrDefault(configuration); + switch (connectionReuseStrategy) { + case LIFO: + connMgrBuilder.setConnPoolPolicy(PoolReusePolicy.LIFO); + break; + case FIFO: + connMgrBuilder.setConnPoolPolicy(PoolReusePolicy.FIFO); + break; + default: + throw new ClientMisconfigurationException("Unknown connection reuse strategy: " + connectionReuseStrategy); + } + LOG.debug("Connection reuse strategy: {}", connectionReuseStrategy); + + connMgrBuilder.setDefaultConnectionConfig(createConnectionConfig(configuration)); + connMgrBuilder.setMaxConnTotal(Integer.MAX_VALUE); // as we do not know how many routes we will have + ClientConfigProperties.HTTP_MAX_OPEN_CONNECTIONS.applyIfSet(configuration, connMgrBuilder::setMaxConnPerRoute); + + int networkBufferSize = ClientConfigProperties.CLIENT_NETWORK_BUFFER_SIZE.getOrDefault(configuration); + MeteredManagedHttpClientConnectionFactory connectionFactory = new MeteredManagedHttpClientConnectionFactory( + Http1Config.custom() + .setBufferSize(networkBufferSize) + .build(), + CharCodingConfig.DEFAULT, + DefaultHttpResponseParserFactory.INSTANCE); + + connMgrBuilder.setConnectionFactory(connectionFactory); + connMgrBuilder.setSSLSocketFactory(sslConnectionSocketFactory); + connMgrBuilder.setDefaultSocketConfig(socketConfig); + PoolingHttpClientConnectionManager phccm = connMgrBuilder.build(); + poolControl = phccm; + if (metricsRegistry != null) { + try { + String mGroupName = ClientConfigProperties.METRICS_GROUP_NAME.getOrDefault(configuration); + Class micrometerLoader = getClass().getClassLoader().loadClass("com.clickhouse.client.api.metrics.MicrometerLoader"); + Method applyMethod = micrometerLoader.getDeclaredMethod("applyPoolingMetricsBinder", Object.class, String.class, PoolingHttpClientConnectionManager.class); + applyMethod.invoke(micrometerLoader, metricsRegistry, mGroupName, phccm); + + applyMethod = micrometerLoader.getDeclaredMethod("applyConnectionMetricsBinder", Object.class, String.class, MeteredManagedHttpClientConnectionFactory.class); + applyMethod.invoke(micrometerLoader, metricsRegistry, mGroupName, connectionFactory); + } catch (Exception e) { + LOG.error("Failed to register metrics", e); + } + } + return phccm; + } + + public CloseableHttpClient createHttpClient(boolean initSslContext, Map configuration) { + // Top Level builders + HttpClientBuilder clientBuilder = HttpClientBuilder.create(); + // An application-supplied SSLContext is used as is; otherwise one is built from the configured + // trust/key material. Server hostname verification below still applies via the SSL mode. + SSLContext sslContext = null; + if (initSslContext) { + Object customSSLContext = configuration.get(ClientConfigProperties.SSL_CONTEXT.getKey()); + if (customSSLContext == null) { + sslContext = createSSLContext(configuration); + } else if (customSSLContext instanceof SSLContext) { + sslContext = (SSLContext) customSSLContext; + } else { + throw new ClientMisconfigurationException("'" + ClientConfigProperties.SSL_CONTEXT.getKey() + + "' must be a javax.net.ssl.SSLContext instance but was " + + customSSLContext.getClass().getName() + "; supply it via Client.Builder.setSSLContext(...)"); + } + } + LayeredConnectionSocketFactory sslConnectionSocketFactory; + if (sslContext != null) { + String socketSNI = (String)configuration.get(ClientConfigProperties.SSL_SOCKET_SNI.getKey()); + SSLMode sslMode = ClientConfigProperties.SSL_MODE.getOrDefault(configuration); + // Trust and VerifyCa skip hostname verification. The same applies when a custom SNI is + // set because the connection hostname will not match the certificate. + boolean trustAllHostnames = sslMode == SSLMode.TRUST || sslMode == SSLMode.VERIFY_CA; + boolean hasSNI = socketSNI != null && !socketSNI.trim().isEmpty(); + // ssl_cipher_suites is parsed and sanitized in ClientConfigProperties (blank tokens dropped, + // names trimmed); an unset or empty list means "no restriction" so the transport defaults apply. + List cipherSuites = ClientConfigProperties.SSL_CIPHER_SUITES.getOrDefault(configuration); + String[] enabledCipherSuites = cipherSuites == null || cipherSuites.isEmpty() ? null + : cipherSuites.toArray(new String[0]); + if (hasSNI || trustAllHostnames || enabledCipherSuites != null) { + // Skip hostname verification only for trust-all modes or when a custom SNI is used (the + // connection hostname would not match the certificate); otherwise a null verifier makes the + // factory fall back to the JDK/HttpClient default verifier, keeping STRICT verification. + // java:S5527 - the permissive verifier is applied only for SSLMode.TRUST/VERIFY_CA (where the + // user has explicitly opted out of hostname verification) or a custom SNI; STRICT keeps the + // default verifying behaviour, so secure-by-default hostname verification is preserved. + @SuppressWarnings("java:S5527") + HostnameVerifier hostnameVerifier = trustAllHostnames || hasSNI ? (hostname, session) -> true : null; + sslConnectionSocketFactory = new CustomSSLConnectionFactory(socketSNI, sslContext, hostnameVerifier, + enabledCipherSuites); + } else { + sslConnectionSocketFactory = new SSLConnectionSocketFactory(sslContext); + } + } else { + sslConnectionSocketFactory = new DummySSLConnectionSocketFactory(); + } + // Socket configuration + SocketConfig.Builder soCfgBuilder = SocketConfig.custom(); + ClientConfigProperties.SOCKET_OPERATION_TIMEOUT.applyIfSet(configuration, + (t) -> soCfgBuilder.setSoTimeout(t, TimeUnit.MILLISECONDS)); + + ClientConfigProperties.SOCKET_RCVBUF_OPT.applyIfSet(configuration, + soCfgBuilder::setRcvBufSize); + + ClientConfigProperties.SOCKET_SNDBUF_OPT.applyIfSet(configuration, + soCfgBuilder::setSndBufSize); + + ClientConfigProperties.SOCKET_LINGER_OPT.applyIfSet(configuration, + (v) -> soCfgBuilder.setSoLinger(v, TimeUnit.SECONDS)); + + ClientConfigProperties.SOCKET_TCP_NO_DELAY_OPT.applyIfSet(configuration, + soCfgBuilder::setTcpNoDelay); + // Proxy + String proxyHost = (String) configuration.get(ClientConfigProperties.PROXY_HOST.getKey()); + Integer proxyPort = (Integer) configuration.get(ClientConfigProperties.PROXY_PORT.getKey()); + HttpHost proxy = null; + if (proxyHost != null && proxyPort != null) { + proxy = new HttpHost(proxyHost, proxyPort); + } + + String proxyTypeVal = (String) configuration.get(ClientConfigProperties.PROXY_TYPE.getKey()); + ProxyType proxyType = proxyTypeVal == null ? null : ProxyType.valueOf(proxyTypeVal); + if (proxyType == ProxyType.HTTP) { + clientBuilder.setProxy(proxy); + String proxyUser = (String) configuration.get(ClientConfigProperties.PROXY_USER.getKey()); + String proxyPassword = (String) configuration.get(ClientConfigProperties.PROXY_PASSWORD.getKey()); + if (proxyUser != null && proxyPassword != null) { + proxyAuthHeaderValue = "Basic " + Base64.getEncoder().encodeToString((proxyUser + ":" + proxyPassword).getBytes(StandardCharsets.UTF_8)); + } + + } else if (proxyType == ProxyType.SOCKS) { + soCfgBuilder.setSocksProxyAddress(new InetSocketAddress(proxyHost, proxyPort)); + } + + boolean disableCookies = !((Boolean)ClientConfigProperties.HTTP_SAVE_COOKIES.getOrDefault(configuration)); + if (disableCookies) { + clientBuilder.disableCookieManagement(); + } + SocketConfig socketConfig = soCfgBuilder.build(); + + // Connection manager + if (ClientConfigProperties.CONNECTION_POOL_ENABLED.getOrDefault(configuration)) { + clientBuilder.setConnectionManager(poolConnectionManager(sslConnectionSocketFactory, socketConfig, configuration)); + } else { + clientBuilder.setConnectionManager(basicConnectionManager(sslConnectionSocketFactory, socketConfig, configuration)); + } + Long keepAliveTimeout = ClientConfigProperties.HTTP_KEEP_ALIVE_TIMEOUT.getOrDefault(configuration); + if (keepAliveTimeout != null && keepAliveTimeout > 0) { + clientBuilder.setKeepAliveStrategy((response, context) -> TimeValue.ofMilliseconds(keepAliveTimeout)); + } + + clientBuilder.disableContentCompression(); // will handle ourselves + + return clientBuilder.build(); + } + + private static final String ERROR_CODE_PREFIX_PATTERN = "%d. DB::Exception:"; + + /** + * Reads status line and if error tries to parse response body to get server error message. + * + * @param httpResponse - HTTP response + * @return exception object with server code + */ + public Exception readError(HttpPost req, ClassicHttpResponse httpResponse) { + final String queryId = getQueryId(httpResponse, req); + int serverCode = getHeaderInt(httpResponse.getFirstHeader(ClickHouseHttpProto.HEADER_EXCEPTION_CODE), 0); + try { + return serverCode > 0 ? readClickHouseError(httpResponse.getEntity(), serverCode, queryId, httpResponse.getCode()) : + readNotClickHouseError(httpResponse.getEntity(), queryId, httpResponse.getCode()); + } catch (Exception e) { + LOG.error("Failed to read error message", e); + String msg = String.format(ERROR_CODE_PREFIX_PATTERN, serverCode) + " (transport error: " + httpResponse.getCode() + ")"; + return new ServerException(serverCode, msg + " (queryId= " + queryId + ")", httpResponse.getCode(), queryId); + } + } + + private ServerException readNotClickHouseError(HttpEntity httpEntity, String queryId, int httpCode) { + + byte[] buffer = new byte[ERROR_BODY_BUFFER_SIZE]; + + String msg = null; + InputStream body = null; + int offset = 0; + for (int i = 0; i < 2; i++) { + try { + if (body == null) { + body = httpEntity.getContent(); + } + int msgLen = body.read(buffer, offset, buffer.length - offset); + if (msgLen > 0) { + msg = new String(buffer, 0, msgLen, StandardCharsets.UTF_8).trim(); + if (msg.isEmpty()) { + msg = ""; + } + } + break; + } catch (ClientException e) { + // Invalid LZ4 Magic + if (body instanceof CompressedBlockInputStream) { + CompressedBlockInputStream stream = (CompressedBlockInputStream) body; + body = stream.getInputStream(); + byte[] lzHeader = stream.getHeaderBuffer(); // Here is read part of original body + offset = Math.min(lzHeader.length, buffer.length); + System.arraycopy(lzHeader, 0, buffer, 0, offset); + continue; + } + throw e; + } catch (Exception e) { + LOG.warn("Failed to read error message (queryId = " + queryId + ")", e); + break; + } + } + + String errormsg = msg == null ? "unknown server error" : msg; + return new ServerException(ServerException.CODE_UNKNOWN, errormsg + " (transport error: " + httpCode +")", httpCode, queryId); + } + + private static ServerException readClickHouseError(HttpEntity httpEntity, int serverCode, String queryId, int httpCode) throws Exception { + InputStream body = httpEntity.getContent(); + byte[] buffer = new byte[ERROR_BODY_BUFFER_SIZE]; + byte[] lookUpStr = String.format(ERROR_CODE_PREFIX_PATTERN, serverCode).getBytes(StandardCharsets.UTF_8); + StringBuilder msgBuilder = new StringBuilder(); + boolean found = false; + while (true) { + int rBytes = -1; + try { + rBytes = body.read(buffer); + } catch (ClientException e) { + // Invalid LZ4 Magic + if (body instanceof CompressedBlockInputStream) { + CompressedBlockInputStream stream = (CompressedBlockInputStream) body; + body = stream.getInputStream(); + byte[] headerBuffer = stream.getHeaderBuffer(); + System.arraycopy(headerBuffer, 0, buffer, 0, headerBuffer.length); + rBytes = headerBuffer.length; + } + } + if (rBytes == -1) { + break; + } + + for (int i = 0; i < rBytes; i++) { + if (buffer[i] == lookUpStr[0]) { + found = true; + for (int j = 1; j < Math.min(rBytes - i, lookUpStr.length); j++) { + if (buffer[i + j] != lookUpStr[j]) { + found = false; + break; + } + } + if (found) { + msgBuilder.append(new String(buffer, i, rBytes - i, StandardCharsets.UTF_8)); + break; + } + } + } + + if (found) { + break; + } + } + + while (true) { + int rBytes = body.read(buffer); + if (rBytes == -1) { + break; + } + msgBuilder.append(new String(buffer, 0, rBytes, StandardCharsets.UTF_8)); + } + + String msg = msgBuilder.toString().replaceAll("\\s+", " ").replaceAll("\\\\n", " ") + .replaceAll("\\\\/", "/"); + if (msg.trim().isEmpty()) { + msg = String.format(ERROR_CODE_PREFIX_PATTERN, serverCode) + " (transport error: " + httpCode + ")"; + } + return new ServerException(serverCode, "Code: " + msg + " (queryId= " + queryId + ")", httpCode, queryId); + } + + private static final long POOL_VENT_TIMEOUT = 10000L; + private final AtomicLong timeToPoolVent = new AtomicLong(0); + + private void doPoolVent() { + if (poolControl != null && timeToPoolVent.get() < System.currentTimeMillis()) { + timeToPoolVent.set(System.currentTimeMillis() + POOL_VENT_TIMEOUT); + poolControl.closeExpired(); + } + } + + private HttpContext createRequestHttpContext(Map requestConfig) { + HttpClientContext context = HttpClientContext.create(); + Number responseTimeout = ClientConfigProperties.SOCKET_OPERATION_TIMEOUT.getOrDefault(requestConfig); + Number connectionReqTimeout = ClientConfigProperties.CONNECTION_REQUEST_TIMEOUT.getOrDefault(requestConfig); + RequestConfig reqHttpConf = RequestConfig.custom() + .setResponseTimeout(responseTimeout.longValue(), TimeUnit.MILLISECONDS) + .setConnectionRequestTimeout(connectionReqTimeout.longValue(), TimeUnit.MILLISECONDS) + .build(); + context.setRequestConfig(reqHttpConf); + + return context; + } + + private URI createRequestURI(Endpoint server, Map requestConfig, boolean isMultipartRequest) { + URI uri; + try { + URIBuilder uriBuilder = new URIBuilder(server.getURI()); + addRequestParams(requestConfig, uriBuilder::addParameter); + + if (!isMultipartRequest) { + addStatementParams(requestConfig, uriBuilder::addParameter); + } else { + uriBuilder.removeParameter(ClickHouseHttpProto.QPARAM_DECOMPRESS); // multipart request doesn't support compression yet + } + + uri = uriBuilder.optimize().build(); + } catch (URISyntaxException e) { + throw new RuntimeException(e); + } + return uri; + } + + private HttpPost createPostRequest(URI uri, Map requestConfig) { + HttpPost req = new HttpPost(uri); +// req.setVersion(new ProtocolVersion("HTTP", 1, 0)); // to disable chunk transfer encoding + addHeaders(req, requestConfig); + return req; + } + + private static final class TransportRequestImpl implements TransportRequest { + private final HttpPost delegate; + private final Map config; + private final AtomicBoolean cancelled = new AtomicBoolean(false); + + TransportRequestImpl(HttpPost delegate, Map config) { + this.delegate = delegate; + this.config = config; + } + + @Override + public boolean cancel() { + cancelled.set(true); + if (delegate.isCancelled()) { + return true; + } + return delegate.cancel(); + } + + @Override + public boolean isCancelled() { + return cancelled.get(); + } + + @Override + public Map getConfig() { + return config; + } + + @Override + @SuppressWarnings("unchecked") + public T getDelegate() { + return (T) delegate; + } + } + + public TransportRequest createRequest(Endpoint server, Map requestConfig, + String body) { + boolean useMultipart = ClientConfigProperties.HTTP_SEND_PARAMS_IN_BODY.getOrDefault(requestConfig) && + requestConfig.containsKey(HttpAPIClientHelper.KEY_STATEMENT_PARAMS); + + // create configuration dependent objects + final URI uri = createRequestURI(server, requestConfig, useMultipart); + final HttpPost req = createPostRequest(uri, requestConfig); + + final HttpEntity httpEntity; + if (useMultipart) { + // a multipart body is always sent as-is, so the request must not declare a content encoding - the + // server would fail to decompress the plain body. Removed after addHeaders() to also drop an + // encoding set by the application with `http_header_*`. + req.removeHeaders(HttpHeaders.CONTENT_ENCODING); + + MultipartEntityBuilder multipartEntityBuilder = MultipartEntityBuilder.create(); + addStatementParams(requestConfig, multipartEntityBuilder::addTextBody); + multipartEntityBuilder.addTextBody(ClickHouseHttpProto.QPARAM_QUERY_STMT, body); + + httpEntity = multipartEntityBuilder.build(); + req.setHeader(HttpHeaders.CONTENT_TYPE, httpEntity.getContentType()); // set proper content type with generated boundary value + } else { + try { + final String contentEncoding = req.containsHeader(HttpHeaders.CONTENT_ENCODING) ? req.getHeader(HttpHeaders.CONTENT_ENCODING).getValue() : null; + httpEntity = new ByteArrayEntity(body.getBytes(StandardCharsets.UTF_8.name()), CONTENT_TYPE, contentEncoding); + } catch (UnsupportedEncodingException | ProtocolException e) { + throw new ClientException("failed to create request body entity", e); + } + } + // adjust configuration + if (useMultipart) { + req.setEntity(httpEntity); // multipart doesn't support compression right now + } else { + req.setEntity(wrapRequestEntity(httpEntity, requestConfig)); + } + return new TransportRequestImpl(req, requestConfig); + } + + + private static final class TransportResponseImpl implements TransportResponse { + + private final ClassicHttpResponse delegate; + + TransportResponseImpl(ClassicHttpResponse delegate) { + this.delegate = delegate; + } + + @Override + public ClickHouseFormat getDataFormat() { + Header formatHeader = delegate.getFirstHeader(ClickHouseHttpProto.HEADER_FORMAT); + return formatHeader == null ? null : ClickHouseFormat.valueOf(formatHeader.getValue()); + } + + @Override + public String getSummaryJson() { + return HttpAPIClientHelper.getHeaderVal(delegate + .getFirstHeader(ClickHouseHttpProto.HEADER_SRV_SUMMARY), "{}"); + } + + @Override + public String getQueryId() { + return HttpAPIClientHelper.getHeaderVal(delegate + .getFirstHeader(ClickHouseHttpProto.HEADER_QUERY_ID), null); + } + + @Override + @SuppressWarnings("unchecked") + public T getDelegate() { + return (T) delegate; + } + + @Override + public Map getHeaders() { + return HttpAPIClientHelper.collectResponseHeaders(delegate); + } + + @Override + public void close() throws IOException { + delegate.close(); + } + + @Override + public InputStream createDataInputStream() { + try { + return delegate.getEntity().getContent(); + } catch (Exception e) { + throw new ClientException("Failed to construct input stream", e); + } + } + } + + /** + * Executes a single transport request and records it as a child span of the given operation + * span, so a retried operation reports one request span per attempt. The request itself is + * executed by {@link #doExecuteRequest(TransportRequest, Span)}, which stays the single place + * where a request is actually executed and which records the HTTP status of the response. + * + *

+ * When no recorder is registered the request is executed by + * {@link #executeRequest(TransportRequest)}, so a client that does not observe spans keeps + * exactly the behaviour it had before and pays nothing for the span path. + * + * @param transportRequest - request to execute + * @param operationSpan - span of the operation this request is made for + * @return transport response + * @throws Exception when the request could not be completed + */ + public TransportResponse executeRequest(TransportRequest transportRequest, Span operationSpan) throws Exception { + if (spanRecorder == DefaultSpanRecorder.NOOP) { + return executeRequest(transportRequest); + } + + final Span requestSpan = startRequestSpan(operationSpan, transportRequest.getDelegate()); + try { + return doExecuteRequest(transportRequest, requestSpan); + } catch (Exception e) { + spanRecorder.recordRequestFailure(requestSpan, e); + throw e; + } finally { + requestSpan.end(); + } + } + + private Span startRequestSpan(Span operationSpan, HttpPost req) { + final URIAuthority authority = req.getAuthority(); + Span span = spanRecorder.startRequestSpan(operationSpan, + authority == null ? null : authority.getHostName(), + authority == null ? -1 : authority.getPort()); + return span == null ? DefaultSpanRecorder.NOOP_SPAN : span; + } + + public TransportResponse executeRequest(TransportRequest transportRequest) throws Exception { + return doExecuteRequest(transportRequest, DefaultSpanRecorder.NOOP_SPAN); + } + + /** + * Executes a single transport request and records the HTTP status on the given request span as + * soon as a response is received - so the status is reported for every response, also for the + * ones this method maps onto an exception that does not carry it (for example {@code 502} and + * {@code 503} onto {@link ConnectException}). + * + * @param transportRequest - request to execute + * @param requestSpan - span of this request; {@link DefaultSpanRecorder#NOOP_SPAN} when the + * request is not recorded + * @return transport response + * @throws Exception when the request could not be completed + */ + private TransportResponse doExecuteRequest(TransportRequest transportRequest, Span requestSpan) throws Exception { + + final Map requestConfig = transportRequest.getConfig(); + final HttpPost req = transportRequest.getDelegate(); + + doPoolVent(); + + ClassicHttpResponse httpResponse = null; + boolean closeResponse = true; + HttpContext context = createRequestHttpContext(requestConfig); + try { + httpResponse = httpClient.executeOpen(null, req, context); + if (requestSpan != DefaultSpanRecorder.NOOP_SPAN) { + // nothing to report when this request is not recorded - and a recorder is never handed a + // span it did not create + spanRecorder.recordHttpStatus(requestSpan, httpResponse.getCode()); + } + + httpResponse.setEntity(wrapResponseEntity(httpResponse.getEntity(), + httpResponse.getCode(), + requestConfig)); + + if (httpResponse.containsHeader(ClickHouseHttpProto.HEADER_EXCEPTION_CODE)) { + throw readError(req, httpResponse); + } + + int statusCode = httpResponse.getCode(); + switch (statusCode) { + case HttpStatus.SC_OK: + closeResponse = false; + return new TransportResponseImpl(httpResponse); + case HttpStatus.SC_PROXY_AUTHENTICATION_REQUIRED: + throw new ClientMisconfigurationException("Proxy authentication required. Please check your proxy settings."); + case HttpStatus.SC_BAD_GATEWAY: + throw new ConnectException("Server returned '502 Bad gateway'. Check network and proxy settings."); + case HttpStatus.SC_SERVICE_UNAVAILABLE: + throw new ConnectException("Server returned '503 Service Unavailable'. Check network settings."); + case HttpStatus.SC_BAD_REQUEST: + case HttpStatus.SC_UNAUTHORIZED: + case HttpStatus.SC_FORBIDDEN: + case HttpStatus.SC_SERVER_ERROR: + case HttpStatus.SC_NOT_FOUND: + // ClickHouse usually uses SC_BAD_REQUEST and SC_SERVER_ERROR to return error. + // SC_UNAUTHORIZED, SC_FORBIDDEN is for authentication + // SC_NOT_FOUND can be returned by ClickHouse when path doesn't match database, but also by proxy + // others we cannot handle properly + throw readError(req, httpResponse); + default: + // Unknown status code: log it once (the generic exception below carries no server context). + logServerErrorResponse(req, httpResponse); + throw new ClientException("Unexpected result status " + statusCode); + } + } catch (UnknownHostException e) { + LOG.debug("Host '{}' unknown", req.getAuthority()); + throw e; + } catch (ConnectException | NoRouteToHostException e) { + LOG.debug("Failed to connect to '{}': {}", req.getAuthority(), e.getMessage()); + throw e; + } catch (Exception e) { + LOG.debug("Failed to execute request to '{}': {}", req.getAuthority(), e.getMessage(), e); + if (e instanceof RequestFailedException && req.isCancelled()) { + throw new TransportException("Request was cancelled on client side", e, getQueryId(httpResponse, req)); + } + throw e; + } finally { + if (closeResponse) { + ClientUtils.quietClose(httpResponse, LOG); + } + } + } + + public TransportRequest createRequest(Endpoint server, Map requestConfig, IOCallback writeCallback) { + final URI uri = createRequestURI(server, requestConfig, false); + final HttpPost req = createPostRequest(uri, requestConfig); + try { + String contentEncoding = req.containsHeader(HttpHeaders.CONTENT_ENCODING) ? req.getHeader(HttpHeaders.CONTENT_ENCODING).getValue() : null; + req.setEntity(wrapRequestEntity( + new EntityTemplate(-1, CONTENT_TYPE, contentEncoding, writeCallback), + requestConfig)); + } catch (ProtocolException e) { + throw new ClientException("failed to create request body entity", e); + } + + return new TransportRequestImpl(req, requestConfig); + } + + private String getQueryId(HttpResponse httpResponse, HttpPost httpRequest) { + final Header serverQueryIdHeader = httpResponse == null ? null : httpResponse.getFirstHeader(ClickHouseHttpProto.HEADER_QUERY_ID); + final Header clientQueryIdHeader = httpRequest == null ? null : httpRequest.getFirstHeader(ClickHouseHttpProto.HEADER_QUERY_ID); + final Header queryHeader = Stream.of(serverQueryIdHeader, clientQueryIdHeader).filter(Objects::nonNull).findFirst().orElse(null); + return queryHeader == null ? "" : queryHeader.getValue(); + } + + /** + * Logs a server error response at WARN (status, query id, authority, exception-code header). The body is + * not logged: {@link #readError} consumes it and it may contain SQL/data. + */ + private void logServerErrorResponse(HttpPost req, ClassicHttpResponse httpResponse) { + // Null-safe: the error path must never let its own logger throw and mask the real failure. + if (req == null || httpResponse == null) { + return; + } + final Header exceptionCodeHeader = httpResponse.getFirstHeader(ClickHouseHttpProto.HEADER_EXCEPTION_CODE); + LOG.warn("Server returned error response: status={}, queryId='{}', authority='{}', {}={}", + httpResponse.getCode(), + getQueryId(httpResponse, req), + req.getAuthority(), + ClickHouseHttpProto.HEADER_EXCEPTION_CODE, + exceptionCodeHeader == null ? "" : exceptionCodeHeader.getValue()); + } + + private static final ContentType CONTENT_TYPE = ContentType.create(ContentType.TEXT_PLAIN.getMimeType(), "UTF-8"); + + private void addHeaders(HttpPost req, Map requestConfig) { + setHeader(req, HttpHeaders.CONTENT_TYPE, CONTENT_TYPE.getMimeType()); + String socketSNI = (String) requestConfig.get(ClientConfigProperties.SSL_SOCKET_SNI.getKey()); + if (ClientUtils.isNotBlank(socketSNI) && !req.containsHeader(HttpHeaders.HOST)) { + int port = -1; + try { + if (req.getUri() != null) { + port = req.getUri().getPort(); + } + } catch (Exception ignore) { + // ignore + } + String hostHeader = (port > 0 && port != 80 && port != 443) ? socketSNI + ":" + port : socketSNI; + setHeader(req, HttpHeaders.HOST, hostHeader); + } + if (requestConfig.containsKey(ClientConfigProperties.INPUT_OUTPUT_FORMAT.getKey())) { + Object formatObj = requestConfig.get(ClientConfigProperties.INPUT_OUTPUT_FORMAT.getKey()); + if (formatObj != null) { + String formatStr = formatObj instanceof String ? formatObj.toString() : ((ClickHouseFormat)formatObj).name(); + if (ClientUtils.isNotBlank(formatStr)) { + setHeader( + req, + ClickHouseHttpProto.HEADER_FORMAT, + formatStr); + } + } + } + if (requestConfig.containsKey(ClientConfigProperties.QUERY_ID.getKey())) { + setHeader( + req, + ClickHouseHttpProto.HEADER_QUERY_ID, + (String) requestConfig.get(ClientConfigProperties.QUERY_ID.getKey())); + } + setHeader( + req, + ClickHouseHttpProto.HEADER_DATABASE, + ClientConfigProperties.DATABASE.getOrDefault(requestConfig)); + + if (ClientConfigProperties.SSL_AUTH.getOrDefault(requestConfig).booleanValue()) { + setHeader( + req, + ClickHouseHttpProto.HEADER_DB_USER, + ClientConfigProperties.USER.getOrDefault(requestConfig)); + setHeader( + req, + ClickHouseHttpProto.HEADER_SSL_CERT_AUTH, + "on"); + } else if (ClientConfigProperties.HTTP_USE_BASIC_AUTH.getOrDefault(requestConfig).booleanValue()) { + String user = ClientConfigProperties.USER.getOrDefault(requestConfig); + String password = ClientConfigProperties.PASSWORD.getOrDefault(requestConfig); + // Use as-is, no encoding allowed + req.addHeader( + HttpHeaders.AUTHORIZATION, + "Basic " + Base64.getEncoder().encodeToString( + (user + ":" + password).getBytes(StandardCharsets.UTF_8))); + } else { + setHeader( + req, + ClickHouseHttpProto.HEADER_DB_USER, + ClientConfigProperties.USER.getOrDefault(requestConfig)); + setHeader( + req, + ClickHouseHttpProto.HEADER_DB_PASSWORD, + ClientConfigProperties.PASSWORD.getOrDefault(requestConfig)); + } + if (proxyAuthHeaderValue != null) { + req.addHeader( + HttpHeaders.PROXY_AUTHORIZATION, + proxyAuthHeaderValue); + } + + boolean clientCompression = ClientConfigProperties.COMPRESS_CLIENT_REQUEST.getOrDefault(requestConfig); + boolean serverCompression = ClientConfigProperties.COMPRESS_SERVER_RESPONSE.getOrDefault(requestConfig); + boolean useHttpCompression = ClientConfigProperties.USE_HTTP_COMPRESSION.getOrDefault(requestConfig); + boolean appCompressedData = ClientConfigProperties.APP_COMPRESSED_DATA.getOrDefault(requestConfig); + + if (useHttpCompression) { + if (serverCompression) { + setHeader(req, HttpHeaders.ACCEPT_ENCODING, DEFAULT_HTTP_COMPRESSION_ALGO); + } + + if (clientCompression && !appCompressedData) { + setHeader(req, HttpHeaders.CONTENT_ENCODING, DEFAULT_HTTP_COMPRESSION_ALGO); + } + } + + for (String key : requestConfig.keySet()) { + if (key.startsWith(ClientConfigProperties.HTTP_HEADER_PREFIX)) { + Object val = requestConfig.get(key); + if (val != null) { + setHeader( + req, + key.substring(ClientConfigProperties.HTTP_HEADER_PREFIX.length()), + String.valueOf(val)); + } + } + } + + // Special cases + if (req.containsHeader(HttpHeaders.AUTHORIZATION) + && (req.containsHeader(ClickHouseHttpProto.HEADER_DB_USER) || + req.containsHeader(ClickHouseHttpProto.HEADER_DB_PASSWORD))) + { + // user has set auth header for purpose, lets remove ours + req.removeHeaders(ClickHouseHttpProto.HEADER_DB_USER); + req.removeHeaders(ClickHouseHttpProto.HEADER_DB_PASSWORD); + } + + // -- keep last + correctUserAgentHeader(req, requestConfig); + } + + private void addRequestParams(Map requestConfig, BiConsumer consumer) { + if (requestConfig.containsKey(ClientConfigProperties.QUERY_ID.getKey())) { + consumer.accept(ClickHouseHttpProto.QPARAM_QUERY_ID, requestConfig.get(ClientConfigProperties.QUERY_ID.getKey()).toString()); + } + + boolean clientCompression = ClientConfigProperties.COMPRESS_CLIENT_REQUEST.getOrDefault(requestConfig); + boolean serverCompression = ClientConfigProperties.COMPRESS_SERVER_RESPONSE.getOrDefault(requestConfig); + boolean useHttpCompression = ClientConfigProperties.USE_HTTP_COMPRESSION.getOrDefault(requestConfig); + + if (useHttpCompression) { + // enable_http_compression make server react on http header + // for client side compression Content-Encoding should be set + // for server side compression Accept-Encoding should be set + consumer.accept(ClickHouseHttpProto.QPARAM_ENABLE_HTTP_COMPRESSION, "1"); + } else { + if (serverCompression) { + consumer.accept(ClickHouseHttpProto.QPARAM_COMPRESS, "1"); + } + if (clientCompression) { + consumer.accept(ClickHouseHttpProto.QPARAM_DECOMPRESS, "1"); + } + } + + Collection sessionRoles = ClientConfigProperties.SESSION_DB_ROLES.getOrDefault(requestConfig); + if (!(sessionRoles == null || sessionRoles.isEmpty())) { + sessionRoles.forEach(r -> consumer.accept(ClickHouseHttpProto.QPARAM_ROLE, r)); + } + + for (String key : requestConfig.keySet()) { + if (key.startsWith(ClientConfigProperties.SERVER_SETTING_PREFIX)) { + Object val = requestConfig.get(key); + if (val != null) { + consumer.accept(key.substring(ClientConfigProperties.SERVER_SETTING_PREFIX.length()), String.valueOf(requestConfig.get(key))); + } + } + } + } + + private void addStatementParams(Map requestConfig, BiConsumer consumer) { + if (requestConfig.containsKey(KEY_STATEMENT_PARAMS)) { + Map params = (Map) requestConfig.get(KEY_STATEMENT_PARAMS); + params.forEach((k, v) -> consumer.accept("param_" + k, String.valueOf(v))); + } + } + + private HttpEntity wrapRequestEntity(HttpEntity httpEntity, Map requestConfig) { + + boolean clientCompression = ClientConfigProperties.COMPRESS_CLIENT_REQUEST.getOrDefault(requestConfig); + boolean useHttpCompression = ClientConfigProperties.USE_HTTP_COMPRESSION.getOrDefault(requestConfig); + boolean appCompressedData = ClientConfigProperties.APP_COMPRESSED_DATA.getOrDefault(requestConfig); + + if (httpEntity.getContentEncoding() != null && !appCompressedData) { + // http header is set and data is not compressed + return new CompressedEntity(httpEntity, false, CompressorStreamFactory.getSingleton()); + } else if (clientCompression && !appCompressedData) { + int buffSize = ClientConfigProperties.COMPRESSION_LZ4_UNCOMPRESSED_BUF_SIZE.getOrDefault(requestConfig); + CompressionMethod compressMethod = ClientConfigProperties.COMPRESSION_METHOD.getOrDefault(requestConfig); + return new CompressedBlockEntity(httpEntity, useHttpCompression, false, true, + buffSize, false, lz4Factory, compressMethod); + } else { + return httpEntity; + } + } + + private HttpEntity wrapResponseEntity(HttpEntity httpEntity, int httpStatus, Map requestConfig) { + boolean serverCompression = ClientConfigProperties.COMPRESS_SERVER_RESPONSE.getOrDefault(requestConfig); + boolean useHttpCompression = ClientConfigProperties.USE_HTTP_COMPRESSION.getOrDefault(requestConfig); + + if (httpEntity.getContentEncoding() != null) { + // http compressed response + return new CompressedEntity(httpEntity, true, CompressorStreamFactory.getSingleton()); + } + + // data compression + if (serverCompression && !(httpStatus == HttpStatus.SC_FORBIDDEN || httpStatus == HttpStatus.SC_UNAUTHORIZED)) { + int buffSize = ClientConfigProperties.COMPRESSION_LZ4_UNCOMPRESSED_BUF_SIZE.getOrDefault(requestConfig); + return new CompressedBlockEntity(httpEntity, useHttpCompression, true, false, buffSize, true, lz4Factory, + ClientConfigProperties.COMPRESSION_METHOD.getDefObjVal()); + } + + return httpEntity; + } + + public static int getHeaderInt(Header header, int defaultValue) { + return getHeaderVal(header, defaultValue, Integer::parseInt); + } + + private static final Set RESPONSE_HEADER_WHITELIST = new HashSet<>(Arrays.asList( + ClickHouseHttpProto.HEADER_QUERY_ID, + ClickHouseHttpProto.HEADER_SRV_SUMMARY, + ClickHouseHttpProto.HEADER_SRV_DISPLAY_NAME, + ClickHouseHttpProto.HEADER_DATABASE, + ClickHouseHttpProto.HEADER_DB_USER, + ClickHouseHttpProto.HEADER_TIMEZONE, + ClickHouseHttpProto.HEADER_FORMAT, + ClickHouseHttpProto.HEADER_PROGRESS + )); + + /** + * Collects whitelisted response headers from an HTTP response into a map. + * + * @param response the HTTP response + * @return unmodifiable map of header name to header value for whitelisted headers present in the response + */ + public static Map collectResponseHeaders(ClassicHttpResponse response) { + Map headers = new HashMap<>(); + for (String name : RESPONSE_HEADER_WHITELIST) { + Header header = response.getFirstHeader(name); + if (header != null) { + headers.put(name, header.getValue()); + } + } + return Collections.unmodifiableMap(headers); + } + + public static String getHeaderVal(Header header, String defaultValue) { + return getHeaderVal(header, defaultValue, Function.identity()); + } + + public static T getHeaderVal(Header header, T defaultValue, Function converter) { + if (header == null) { + return defaultValue; + } + + return converter.apply(header.getValue()); + } + + public boolean shouldRetry(Throwable ex, Map requestSettings) { + List retryCauses = ClientConfigProperties.CLIENT_RETRY_ON_FAILURE.getOrDefault(requestSettings); + + if (retryCauses.contains(ClientFaultCause.None)) { + return false; + } + + if (ex instanceof NoHttpResponseException + || ex.getCause() instanceof NoHttpResponseException) { + return retryCauses.contains(ClientFaultCause.NoHttpResponse); + } + + if (ex instanceof ConnectException + || ex instanceof ConnectTimeoutException + || ex.getCause() instanceof ConnectException + || ex.getCause() instanceof ConnectTimeoutException) { + return retryCauses.contains(ClientFaultCause.ConnectTimeout); + } + + if (ex instanceof ConnectionRequestTimeoutException + || ex.getCause() instanceof ConnectionRequestTimeoutException) { + return retryCauses.contains(ClientFaultCause.ConnectionRequestTimeout); + } + + if (ex instanceof SocketTimeoutException + || ex.getCause() instanceof SocketTimeoutException) { + return retryCauses.contains(ClientFaultCause.SocketTimeout); + } + + // there are some db retryable error codes + if (ex instanceof ServerException || ex.getCause() instanceof ServerException) { + ServerException se = (ServerException) (ex instanceof ServerException ? ex : ex.getCause()); + if (se.getCode() == ServerException.EXECUTION_TIMEOUT) { + return retryCauses.contains(ClientFaultCause.ServerTimeoutExceeded); + } + return se.isRetryable() && retryCauses.contains(ClientFaultCause.ServerRetryable); + } + + return false; + } + + // This method wraps some client specific exceptions into specific ClientException or just ClientException + // ClientException will be also wrapped + public RuntimeException wrapException(String message, Exception cause, String queryId) { + // Already-classified exceptions (ClientException, ServerException, ConnectionInitiationException, ...) + // are returned as-is so their specific type is preserved instead of being reboxed as a generic + // ClickHouseException. + if (cause instanceof ClickHouseException) { + return (ClickHouseException) cause; + } + + if (cause instanceof SSLException) { + return new TransportException("SSL Problem", cause, queryId); + } + + if (cause instanceof ConnectionRequestTimeoutException || + cause instanceof NoHttpResponseException || + cause instanceof ConnectTimeoutException || + cause instanceof ConnectException || + cause instanceof UnknownHostException || + cause instanceof NoRouteToHostException) { + ConnectionInitiationException ex = new ConnectionInitiationException(message, cause); + ex.setQueryId(queryId); + return ex; + } + + if (cause instanceof SocketTimeoutException || cause instanceof IOException) { + DataTransferException ex = new DataTransferException(message, cause); + ex.setQueryId(queryId); + return ex; + } + // if we can not identify the exception explicitly we catch as our base exception ClickHouseException + return new ClickHouseException(message, cause, queryId); + } + + private void correctUserAgentHeader(HttpRequest request, Map requestConfig) { + //TODO: implement cache for user-agent + Header userAgentHeader = request.getLastHeader(HttpHeaders.USER_AGENT); + request.removeHeaders(HttpHeaders.USER_AGENT); + + String clientName = ClientConfigProperties.CLIENT_NAME.getOrDefault(requestConfig); + String userAgentValue = defaultUserAgent; + if (userAgentHeader == null && clientName != null && !clientName.isEmpty()) { + userAgentValue = clientName + " " + defaultUserAgent; + } else if (userAgentHeader != null) { + userAgentValue = userAgentHeader.getValue() + " " + defaultUserAgent; + } + request.setHeader(HttpHeaders.USER_AGENT, userAgentValue); + } + + private String buildDefaultUserAgent() { + StringBuilder userAgent = new StringBuilder(); + userAgent.append(Client.CLIENT_USER_AGENT); + + String clientVersion = Client.clientVersion; + + userAgent.append(clientVersion); + + userAgent.append(" ("); + userAgent.append(System.getProperty("os.name")); + userAgent.append("; "); + userAgent.append("jvm:").append(System.getProperty("java.version")); + userAgent.append("; "); + + userAgent.setLength(userAgent.length() - 2); + userAgent.append(')'); + + try { + String httpClientVersion = this.httpClient.getClass().getPackage().getImplementationVersion(); + if (Objects.equals(this.httpClient.getClass().getPackage().getImplementationTitle(), this.getClass().getPackage().getImplementationTitle())) { + // shaded jar - all packages have same implementation title + httpClientVersion = "unknown"; + try (InputStream in = Thread.currentThread().getContextClassLoader().getResourceAsStream("client-v2-version.properties")) { + Properties p = new Properties(); + p.load(in); + + String tmp = p.getProperty("apache.http.client.version"); + if (tmp != null && !tmp.isEmpty() && !tmp.equals("${apache.httpclient.version}")) { + httpClientVersion = tmp; + } + } catch (Exception e) { + LOG.debug("Failed to read HTTP client version from client-v2-version.properties", e); + } + } + userAgent.append(" ") + .append("Apache-HttpClient") + .append('/') + .append(httpClientVersion); + } catch (Exception e) { + LOG.warn("failed to construct http client version string", e); + } + return userAgent.toString(); + } + + public void close() { + httpClient.close(CloseMode.IMMEDIATE); + } + + private static void setHeader(HttpRequest req, String headerName, + String value) + { + if (value == null) { + return; + } + + if (value.trim().isEmpty()) { + return; + } + if (PATTERN_HEADER_VALUE_ASCII.matcher(value).matches()) { + req.setHeader(headerName, value); + } else { + try { + req.setHeader( + headerName + "*", + "UTF-8''" + URLEncoder.encode(value, StandardCharsets.UTF_8.name())); + } catch (UnsupportedEncodingException e) { + throw new ClientException("Failed to convert string to UTF8" , e); + } + } + } + + /** + * This factory is used only when no ssl connections are required (no https endpoints). + * Internally http client would create factory and spend time if no supplied. + */ + private static class DummySSLConnectionSocketFactory implements LayeredConnectionSocketFactory { + @Override + public Socket createLayeredSocket(Socket socket, String target, int port, HttpContext context) throws IOException { + return null; + } + + @Override + public Socket createSocket(HttpContext context) throws IOException { + return null; + } + + @Override + public Socket connectSocket(TimeValue connectTimeout, Socket socket, HttpHost host, InetSocketAddress remoteAddress, InetSocketAddress localAddress, HttpContext context) throws IOException { + return null; + } + } + + public static class MeteredManagedHttpClientConnectionFactory extends ManagedHttpClientConnectionFactory { + public MeteredManagedHttpClientConnectionFactory(Http1Config http1Config, CharCodingConfig charCodingConfig, DefaultHttpResponseParserFactory defaultHttpResponseParserFactory) { + super(http1Config, charCodingConfig, defaultHttpResponseParserFactory); + } + + ConcurrentLinkedQueue times = new ConcurrentLinkedQueue<>(); + + + @Override + public ManagedHttpClientConnection createConnection(Socket socket) throws IOException { + long startT = System.currentTimeMillis(); + try { + return super.createConnection(socket); + } finally { + long endT = System.currentTimeMillis(); + times.add(endT - startT); + } + } + + public long getTime() { + int count = times.size(); + long runningAverage = 0; + for (int i = 0; i < count; i++) { + Long t = times.poll(); + if (t != null) { + runningAverage += t; + } + } + + return count > 0 ? runningAverage / count : 0; + } + } + + public static class CustomSSLConnectionFactory extends SSLConnectionSocketFactory { + + private final SNIHostName defaultSNI; + + // Retained for backward compatibility; delegates with no cipher-suite restriction (transport defaults). + public CustomSSLConnectionFactory(String defaultSNI, SSLContext sslContext, HostnameVerifier hostnameVerifier) { + this(defaultSNI, sslContext, hostnameVerifier, null); + } + + public CustomSSLConnectionFactory(String defaultSNI, SSLContext sslContext, HostnameVerifier hostnameVerifier, + String[] supportedCipherSuites) { + super(sslContext, null /* supportedProtocols */, supportedCipherSuites, hostnameVerifier); + this.defaultSNI = defaultSNI == null || defaultSNI.trim().isEmpty() ? null : new SNIHostName(defaultSNI); + } + + @Override + protected void prepareSocket(SSLSocket socket, HttpContext context) throws IOException { + super.prepareSocket(socket, context); + + if (defaultSNI != null) { + SSLParameters sslParams = socket.getSSLParameters(); + sslParams.setServerNames(Collections.singletonList(defaultSNI)); + socket.setSSLParameters(sslParams); + } + } + } +} From e07a08158ca724ceef213185becc48083499a312 Mon Sep 17 00:00:00 2001 From: guruprasad Date: Tue, 20 Oct 2026 04:17:49 +0530 Subject: [PATCH 2/2] Reject unsupported header values instead of sending them under a renamed header. --- .../client/api/internal/HttpAPIClientHelper.java | 11 ++++------- 1 file changed, 4 insertions(+), 7 deletions(-) diff --git a/client-v2/src/main/java/com/clickhouse/client/api/internal/HttpAPIClientHelper.java b/client-v2/src/main/java/com/clickhouse/client/api/internal/HttpAPIClientHelper.java index fc942bf0d..422fc79ea 100644 --- a/client-v2/src/main/java/com/clickhouse/client/api/internal/HttpAPIClientHelper.java +++ b/client-v2/src/main/java/com/clickhouse/client/api/internal/HttpAPIClientHelper.java @@ -1264,13 +1264,10 @@ private static void setHeader(HttpRequest req, String headerName, if (PATTERN_HEADER_VALUE_ASCII.matcher(value).matches()) { req.setHeader(headerName, value); } else { - try { - req.setHeader( - headerName + "*", - "UTF-8''" + URLEncoder.encode(value, StandardCharsets.UTF_8.name())); - } catch (UnsupportedEncodingException e) { - throw new ClientException("Failed to convert string to UTF8" , e); - } + throw new ClientException( + "HTTP header '" + headerName + + "' contains an unsupported value. Only printable ASCII " + + "characters with internal spaces or tabs are supported."); } }