diff --git a/fe/fe-core/src/main/java/org/apache/doris/cloud/CacheHotspotManager.java b/fe/fe-core/src/main/java/org/apache/doris/cloud/CacheHotspotManager.java index bd08f3847d639f..c529d77e61d6b3 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/cloud/CacheHotspotManager.java +++ b/fe/fe-core/src/main/java/org/apache/doris/cloud/CacheHotspotManager.java @@ -43,6 +43,7 @@ import org.apache.doris.common.util.MasterDaemon; import org.apache.doris.common.util.NetUtils; import org.apache.doris.common.util.TimeUtils; +import org.apache.doris.httpv2.client.InternalHttpClientProvider; import org.apache.doris.httpv2.rest.manager.HttpUtils; import org.apache.doris.metric.MetricRepo; import org.apache.doris.nereids.trees.plans.commands.CancelWarmUpJobCommand; @@ -1102,7 +1103,8 @@ private Map collectAndAggregate() { String url = "http://" + NetUtils.getHostPortInAccessibleFormat(be.getHost(), be.getHttpPort()) + "/api/warmup_event_driven_stats"; - String json = HttpUtils.doGet(url, authHeaders, 5000); + String json = HttpUtils.doInternalGet( + url, authHeaders, 5000, InternalHttpClientProvider.Target.BE); return Pair.of(cluster, json); }); } diff --git a/fe/fe-core/src/main/java/org/apache/doris/common/proc/ReplicasProcNode.java b/fe/fe-core/src/main/java/org/apache/doris/common/proc/ReplicasProcNode.java index fb4d5cbd5d42bd..de239a35972efa 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/common/proc/ReplicasProcNode.java +++ b/fe/fe-core/src/main/java/org/apache/doris/common/proc/ReplicasProcNode.java @@ -30,6 +30,8 @@ import org.apache.doris.common.Config; import org.apache.doris.common.util.NetUtils; import org.apache.doris.common.util.TimeUtils; +import org.apache.doris.httpv2.client.InternalHttpClientProvider; +import org.apache.doris.httpv2.client.InternalHttpClientProviderFactory; import org.apache.doris.statistics.query.QueryStatsUtil; import org.apache.doris.system.Backend; @@ -111,8 +113,12 @@ public ProcResult fetchResult() throws AnalysisException { String host = (be == null ? Backend.DUMMY_IP : be.getHost()); int port = (be == null ? 0 : be.getHttpPort()); String hostPort = NetUtils.getHostPortInAccessibleFormat(host, port); - String metaUrl = String.format("http://" + hostPort + "/api/meta/header/%d", tabletId); - String compactionUrl = String.format("http://" + hostPort + "/api/compaction/show?tablet_id=%d", tabletId); + String metaUrl = InternalHttpClientProviderFactory.getProvider().normalizeInternalUrl( + String.format("http://" + hostPort + "/api/meta/header/%d", tabletId), + InternalHttpClientProvider.Target.BE); + String compactionUrl = InternalHttpClientProviderFactory.getProvider().normalizeInternalUrl( + String.format("http://" + hostPort + "/api/compaction/show?tablet_id=%d", tabletId), + InternalHttpClientProvider.Target.BE); String path = ""; if (be != null) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/common/proc/TabletsProcDir.java b/fe/fe-core/src/main/java/org/apache/doris/common/proc/TabletsProcDir.java index 676942d0b1b521..119ce6ba922ff7 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/common/proc/TabletsProcDir.java +++ b/fe/fe-core/src/main/java/org/apache/doris/common/proc/TabletsProcDir.java @@ -34,6 +34,8 @@ import org.apache.doris.common.util.ListComparator; import org.apache.doris.common.util.NetUtils; import org.apache.doris.common.util.TimeUtils; +import org.apache.doris.httpv2.client.InternalHttpClientProvider; +import org.apache.doris.httpv2.client.InternalHttpClientProviderFactory; import org.apache.doris.qe.ConnectContext; import org.apache.doris.statistics.query.QueryStatsUtil; import org.apache.doris.system.Backend; @@ -214,10 +216,13 @@ public List> fetchComparableResult(long version, long backendId String host = (be == null ? Backend.DUMMY_IP : be.getHost()); int port = (be == null ? 0 : be.getHttpPort()); String hostPort = NetUtils.getHostPortInAccessibleFormat(host, port); - String metaUrl = String.format("http://" + hostPort + "/api/meta/header/%d", tabletId); + String metaUrl = InternalHttpClientProviderFactory.getProvider().normalizeInternalUrl( + String.format("http://" + hostPort + "/api/meta/header/%d", tabletId), + InternalHttpClientProvider.Target.BE); tabletInfo.add(metaUrl); - String compactionUrl = String.format( - "http://" + hostPort + "/api/compaction/show?tablet_id=%d", tabletId); + String compactionUrl = InternalHttpClientProviderFactory.getProvider().normalizeInternalUrl( + String.format("http://" + hostPort + "/api/compaction/show?tablet_id=%d", tabletId), + InternalHttpClientProvider.Target.BE); tabletInfo.add(compactionUrl); tabletInfo.add(tablet.getCooldownReplicaId()); if (replica.getCooldownMetaId() == null) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/common/util/HttpURLUtil.java b/fe/fe-core/src/main/java/org/apache/doris/common/util/HttpURLUtil.java index f1fa2f87d4012a..5110e29c3ab43e 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/common/util/HttpURLUtil.java +++ b/fe/fe-core/src/main/java/org/apache/doris/common/util/HttpURLUtil.java @@ -20,32 +20,25 @@ import org.apache.doris.catalog.Env; import org.apache.doris.cloud.security.SecurityChecker; import org.apache.doris.common.Config; +import org.apache.doris.httpv2.client.InternalHttpClientProvider; +import org.apache.doris.httpv2.client.InternalHttpClientProviderFactory; import org.apache.doris.httpv2.meta.MetaBaseAction; import org.apache.doris.system.SystemInfoService.HostInfo; import com.google.common.base.Strings; import com.google.common.collect.Maps; -import org.apache.http.conn.ssl.NoopHostnameVerifier; import java.io.IOException; import java.net.HttpURLConnection; -import java.net.URL; import java.util.Map; -import javax.net.ssl.HttpsURLConnection; public class HttpURLUtil { public static HttpURLConnection getConnectionWithNodeIdent(String request) throws IOException { try { SecurityChecker.getInstance().startSSRFChecking(request); - URL url = new URL(request); - HttpURLConnection conn = (HttpURLConnection) url.openConnection(); - - if (conn instanceof HttpsURLConnection && Config.enable_https) { - HttpsURLConnection httpsConn = (HttpsURLConnection) conn; - httpsConn.setSSLSocketFactory(InternalHttpsUtils.getSslContext().getSocketFactory()); - httpsConn.setHostnameVerifier(NoopHostnameVerifier.INSTANCE); - } + HttpURLConnection conn = InternalHttpClientProviderFactory.getProvider() + .openConnection(request, InternalHttpClientProvider.Target.FE); // Must use Env.getServingEnv() instead of getCurrentEnv(), // because here we need to obtain selfNode through the official service catalog. @@ -64,6 +57,20 @@ public static HttpURLConnection getConnectionWithNodeIdent(String request) throw } } + public static HttpURLConnection getInternalConnection(String request, + InternalHttpClientProvider.Target target) throws IOException { + InternalHttpClientProvider provider = InternalHttpClientProviderFactory.getProvider(); + String normalizedRequest = provider.normalizeInternalUrl(request, target); + try { + SecurityChecker.getInstance().startSSRFChecking(normalizedRequest); + return provider.openConnection(normalizedRequest, target); + } catch (Exception e) { + throw e instanceof IOException ? (IOException) e : new IOException(e); + } finally { + SecurityChecker.getInstance().stopSSRFChecking(); + } + } + public static Map getNodeIdentHeaders() throws IOException { Map headers = Maps.newHashMap(); // Must use Env.getServingEnv() instead of getCurrentEnv(), @@ -83,15 +90,12 @@ public static int getHttpPort() { } public static String buildInternalFeUrl(String host, String path, String queryParams) { - String protocol = Config.enable_https ? "https" : "http"; - int port = getHttpPort(); - - String url = protocol + "://" + NetUtils.getHostPortInAccessibleFormat(host, port) + path; + String url = "http://" + NetUtils.getHostPortInAccessibleFormat(host, Config.http_port) + path; if (queryParams != null && !queryParams.isEmpty()) { url += "?" + queryParams; } - - return url; + return InternalHttpClientProviderFactory.getProvider() + .normalizeInternalUrl(url, InternalHttpClientProvider.Target.FE); } } diff --git a/fe/fe-core/src/main/java/org/apache/doris/httpv2/client/InternalHttpClientProvider.java b/fe/fe-core/src/main/java/org/apache/doris/httpv2/client/InternalHttpClientProvider.java new file mode 100644 index 00000000000000..99e30fe413dba6 --- /dev/null +++ b/fe/fe-core/src/main/java/org/apache/doris/httpv2/client/InternalHttpClientProvider.java @@ -0,0 +1,47 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package org.apache.doris.httpv2.client; + +import org.apache.http.impl.client.CloseableHttpClient; +import org.springframework.web.client.RestTemplate; + +import java.io.IOException; +import java.net.HttpURLConnection; + +/** + * Supplies clients for trusted, node-to-node HTTP traffic. + * + *

The target identifies whether an endpoint belongs to FE or BE. Implementations may use + * different transports because not every internal service enables TLS on the same port. + */ +public interface InternalHttpClientProvider { + enum Target { + FE, + BE + } + + String normalizeInternalUrl(String url, Target target); + + HttpURLConnection openConnection(String url, Target target) throws IOException; + + /** Returns a provider-owned, process-lifetime client. Callers must not close it. */ + CloseableHttpClient getHttpClient(Target target); + + /** Returns a provider-owned, process-lifetime client. */ + RestTemplate getRestTemplate(Target target); +} diff --git a/fe/fe-core/src/main/java/org/apache/doris/httpv2/client/InternalHttpClientProviderFactory.java b/fe/fe-core/src/main/java/org/apache/doris/httpv2/client/InternalHttpClientProviderFactory.java new file mode 100644 index 00000000000000..84de4187018336 --- /dev/null +++ b/fe/fe-core/src/main/java/org/apache/doris/httpv2/client/InternalHttpClientProviderFactory.java @@ -0,0 +1,48 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package org.apache.doris.httpv2.client; + +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + +import java.util.Iterator; +import java.util.ServiceLoader; + +public final class InternalHttpClientProviderFactory { + private static final Logger LOG = LogManager.getLogger(InternalHttpClientProviderFactory.class); + private static final InternalHttpClientProvider INSTANCE = loadProvider(); + + private InternalHttpClientProviderFactory() { + } + + public static InternalHttpClientProvider getProvider() { + return INSTANCE; + } + + private static InternalHttpClientProvider loadProvider() { + ServiceLoader loader = ServiceLoader.load(InternalHttpClientProvider.class); + Iterator iterator = loader.iterator(); + if (iterator.hasNext()) { + InternalHttpClientProvider provider = iterator.next(); + LOG.info("Using InternalHttpClientProvider implementation: {}", provider.getClass().getName()); + return provider; + } + LOG.info("No custom InternalHttpClientProvider found, using the OSS implementation"); + return new OssInternalHttpClientProvider(); + } +} diff --git a/fe/fe-core/src/main/java/org/apache/doris/httpv2/client/OssInternalHttpClientProvider.java b/fe/fe-core/src/main/java/org/apache/doris/httpv2/client/OssInternalHttpClientProvider.java new file mode 100644 index 00000000000000..91bab8976842ef --- /dev/null +++ b/fe/fe-core/src/main/java/org/apache/doris/httpv2/client/OssInternalHttpClientProvider.java @@ -0,0 +1,150 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package org.apache.doris.httpv2.client; + +import org.apache.doris.common.Config; +import org.apache.doris.common.util.InternalHttpsUtils; +import org.apache.doris.common.util.NetUtils; +import org.apache.doris.tls.server.TlsProtocolSet; + +import org.apache.http.conn.ssl.NoopHostnameVerifier; +import org.apache.http.impl.client.CloseableHttpClient; +import org.apache.http.impl.client.HttpClientBuilder; +import org.springframework.http.client.SimpleClientHttpRequestFactory; +import org.springframework.web.client.RestTemplate; + +import java.io.IOException; +import java.net.HttpURLConnection; +import java.net.URI; +import java.net.URISyntaxException; +import java.net.URL; +import javax.net.ssl.HttpsURLConnection; + +public class OssInternalHttpClientProvider implements InternalHttpClientProvider { + private final CloseableHttpClient httpClient = HttpClientBuilder.create().build(); + private final RestTemplate restTemplate = new RestTemplate(); + private volatile HttpsClients httpsClients; + + @Override + public String normalizeInternalUrl(String url, Target target) { + if (TlsProtocolSet.isHttpTlsActive()) { + throw new UnsupportedOperationException("FE HTTP TLS requires TLS module"); + } + if (!Config.enable_https || target != Target.FE || isHttps(url)) { + return url; + } + return rewriteSchemeAndPort(url, "https", Config.https_port); + } + + @Override + public HttpURLConnection openConnection(String url, Target target) throws IOException { + HttpURLConnection connection = (HttpURLConnection) new URL(normalizeInternalUrl(url, target)).openConnection(); + if (connection instanceof HttpsURLConnection && Config.enable_https && target == Target.FE) { + HttpsURLConnection httpsConnection = (HttpsURLConnection) connection; + httpsConnection.setSSLSocketFactory(InternalHttpsUtils.getSslContext().getSocketFactory()); + httpsConnection.setHostnameVerifier(NoopHostnameVerifier.INSTANCE); + } + return connection; + } + + @Override + public CloseableHttpClient getHttpClient(Target target) { + if (Config.enable_https && target == Target.FE) { + return getHttpsClients().httpClient; + } + return httpClient; + } + + @Override + public RestTemplate getRestTemplate(Target target) { + if (!Config.enable_https || target != Target.FE) { + return restTemplate; + } + return getHttpsClients().restTemplate; + } + + private HttpsClients getHttpsClients() { + HttpsClients clients = httpsClients; + if (clients != null) { + return clients; + } + synchronized (this) { + clients = httpsClients; + if (clients == null) { + clients = new HttpsClients( + InternalHttpsUtils.createValidatedHttpClient(), + new RestTemplate(new InternalHttpsClientHttpRequestFactory())); + httpsClients = clients; + } + } + return clients; + } + + private static boolean isHttps(String url) { + return url != null && url.regionMatches(true, 0, "https://", 0, "https://".length()); + } + + private static String rewriteSchemeAndPort(String url, String scheme, int port) { + try { + URI uri = new URI(url); + if (uri.getHost() == null) { + throw new IllegalArgumentException("Internal HTTP URL has no host: " + url); + } + + StringBuilder rewritten = new StringBuilder(scheme).append("://"); + if (uri.getRawUserInfo() != null) { + rewritten.append(uri.getRawUserInfo()).append('@'); + } + rewritten.append(NetUtils.getHostPortInAccessibleFormat(uri.getHost(), port)); + if (uri.getRawPath() != null) { + rewritten.append(uri.getRawPath()); + } + if (uri.getRawQuery() != null) { + rewritten.append('?').append(uri.getRawQuery()); + } + if (uri.getRawFragment() != null) { + rewritten.append('#').append(uri.getRawFragment()); + } + return rewritten.toString(); + } catch (URISyntaxException e) { + throw new IllegalArgumentException("Invalid internal HTTP URL: " + url, e); + } + } + + private static class InternalHttpsClientHttpRequestFactory extends SimpleClientHttpRequestFactory { + @Override + protected void prepareConnection(HttpURLConnection connection, String httpMethod) throws IOException { + super.prepareConnection(connection, httpMethod); + if (connection instanceof HttpsURLConnection) { + HttpsURLConnection httpsConnection = (HttpsURLConnection) connection; + httpsConnection.setSSLSocketFactory(InternalHttpsUtils.getSslContext().getSocketFactory()); + httpsConnection.setHostnameVerifier(NoopHostnameVerifier.INSTANCE); + } + } + } + + private static class HttpsClients { + private final CloseableHttpClient httpClient; + private final RestTemplate restTemplate; + + private HttpsClients(CloseableHttpClient httpClient, RestTemplate restTemplate) { + this.httpClient = httpClient; + this.restTemplate = restTemplate; + } + } +} diff --git a/fe/fe-core/src/main/java/org/apache/doris/httpv2/controller/BaseController.java b/fe/fe-core/src/main/java/org/apache/doris/httpv2/controller/BaseController.java index 72c549f35a0768..5a73f22c9c0cb1 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/httpv2/controller/BaseController.java +++ b/fe/fe-core/src/main/java/org/apache/doris/httpv2/controller/BaseController.java @@ -30,6 +30,8 @@ import org.apache.doris.datasource.InternalCatalog; import org.apache.doris.httpv2.HttpAuthManager; import org.apache.doris.httpv2.HttpAuthManager.SessionValue; +import org.apache.doris.httpv2.client.InternalHttpClientProvider; +import org.apache.doris.httpv2.client.InternalHttpClientProviderFactory; import org.apache.doris.httpv2.exception.UnauthorizedException; import org.apache.doris.mysql.privilege.PrivPredicate; import org.apache.doris.qe.ConnectContext; @@ -394,13 +396,9 @@ protected long checkLongParam(String strParam) { } protected String getCurrentFrontendURL() { - if (Config.enable_https) { - // this could be the result of redirection. - return "https://" + NetUtils - .getHostPortInAccessibleFormat(FrontendOptions.getLocalHostAddress(), Config.https_port); - } else { - return "http://" + NetUtils - .getHostPortInAccessibleFormat(FrontendOptions.getLocalHostAddress(), Config.http_port); - } + String url = "http://" + NetUtils + .getHostPortInAccessibleFormat(FrontendOptions.getLocalHostAddress(), Config.http_port); + return InternalHttpClientProviderFactory.getProvider().normalizeInternalUrl(url, + InternalHttpClientProvider.Target.FE); } } diff --git a/fe/fe-core/src/main/java/org/apache/doris/httpv2/controller/SessionController.java b/fe/fe-core/src/main/java/org/apache/doris/httpv2/controller/SessionController.java index a753305106a94b..27d3059d86bc65 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/httpv2/controller/SessionController.java +++ b/fe/fe-core/src/main/java/org/apache/doris/httpv2/controller/SessionController.java @@ -21,6 +21,7 @@ import org.apache.doris.catalog.SchemaTable; import org.apache.doris.catalog.Table; import org.apache.doris.common.util.HttpURLUtil; +import org.apache.doris.httpv2.client.InternalHttpClientProvider; import org.apache.doris.httpv2.entity.ResponseBody; import org.apache.doris.httpv2.entity.ResponseEntityBuilder; import org.apache.doris.httpv2.rest.RestBaseController; @@ -119,8 +120,9 @@ private List> getOtherSessionInfo(HttpServletRequest request Frontend frontend) throws IOException { Map header = Maps.newHashMap(); header.put(NodeAction.AUTHORIZATION, request.getHeader(NodeAction.AUTHORIZATION)); - String res = HttpUtils.doGet(HttpURLUtil.buildInternalFeUrl( - frontend.getHost(), "/rest/v1/session", null), header); + String res = HttpUtils.doInternalGet(HttpURLUtil.buildInternalFeUrl( + frontend.getHost(), "/rest/v1/session", null), header, + InternalHttpClientProvider.Target.FE); ObjectMapper objectMapper = new ObjectMapper(); Map jsonMap = objectMapper.readValue(res, new TypeReference>() {}); diff --git a/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/LoadAction.java b/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/LoadAction.java index 065fac754b765d..33b0fb4a9b7eeb 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/LoadAction.java +++ b/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/LoadAction.java @@ -45,6 +45,7 @@ import org.apache.doris.system.BeSelectionPolicy; import org.apache.doris.system.SystemInfoService; import org.apache.doris.thrift.TNetworkAddress; +import org.apache.doris.tls.server.TlsProtocolSet; import com.google.common.base.Strings; import com.google.common.net.HostAndPort; @@ -854,8 +855,7 @@ private RedirectView redirectToStreamLoadForward(HttpServletRequest request, TNe } String redirectUrl = buildRedirectUrlToBackend(request, addr, modifiedPath, redirectQuery); - LOG.info("Redirect stream load forward url: {}, forward_to: {}", - "http://" + addr.getHostname() + ":" + addr.getPort() + modifiedPath, forwardTarget); + LOG.info("Redirect stream load forward url: {}, forward_to: {}", redirectUrl, forwardTarget); RedirectView redirectView = new RedirectView(redirectUrl); redirectView.setContentType("text/html;charset=utf-8"); redirectView.setStatusCode(org.springframework.http.HttpStatus.TEMPORARY_REDIRECT); @@ -898,6 +898,12 @@ private TNetworkAddress handleStreamLoadRedirect(HttpServletRequest request, boo if (!Config.isCloudMode() || !groupCommit || !Config.enable_group_commit_streamload_be_forward) { return selectRedirectBackend(request, groupCommit, tableId); } + if (TlsProtocolSet.isHttpTlsActive()) { + LOG.debug("Group commit stream load BE forward is disabled under HTTP TLS," + + " falling back to a direct group-commit redirect: db={}, tbl={}, label={}", + dbName, tableName, label); + return selectRedirectBackend(request, groupCommit, tableId); + } String cloudClusterName = getCloudClusterName(request); if (Strings.isNullOrEmpty(cloudClusterName)) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/RestBaseController.java b/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/RestBaseController.java index dd77b33d3a4b10..d82958809ac6bf 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/RestBaseController.java +++ b/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/RestBaseController.java @@ -23,8 +23,9 @@ import org.apache.doris.common.FeConstants; import org.apache.doris.common.UserException; import org.apache.doris.common.util.HttpURLUtil; -import org.apache.doris.common.util.InternalHttpsUtils; import org.apache.doris.common.util.NetUtils; +import org.apache.doris.httpv2.client.InternalHttpClientProvider; +import org.apache.doris.httpv2.client.InternalHttpClientProviderFactory; import org.apache.doris.httpv2.controller.BaseController; import org.apache.doris.httpv2.entity.ResponseEntityBuilder; import org.apache.doris.httpv2.exception.UnauthorizedException; @@ -37,7 +38,6 @@ import jakarta.servlet.http.HttpServletRequest; import jakarta.servlet.http.HttpServletResponse; import org.apache.commons.codec.digest.DigestUtils; -import org.apache.http.conn.ssl.NoopHostnameVerifier; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; import org.springframework.http.HttpEntity; @@ -45,7 +45,6 @@ import org.springframework.http.HttpMethod; import org.springframework.http.HttpStatus; import org.springframework.http.ResponseEntity; -import org.springframework.http.client.SimpleClientHttpRequestFactory; import org.springframework.web.client.RestTemplate; import org.springframework.web.servlet.view.RedirectView; @@ -55,13 +54,11 @@ import java.io.FileInputStream; import java.io.IOException; import java.io.OutputStream; -import java.net.HttpURLConnection; import java.net.URI; import java.net.URISyntaxException; import java.util.Collections; import java.util.stream.Collectors; import javax.annotation.Nullable; -import javax.net.ssl.HttpsURLConnection; public class RestBaseController extends BaseController { @@ -103,10 +100,12 @@ protected String buildRedirectUrl(HttpServletRequest request, TNetworkAddress ad return buildRedirectUrl(request.getScheme(), request, addr, requestPath, queryString); } - // BE's stream-load listener never terminates TLS, so BE-bound redirects must stay "http". + // Start from the BE's ordinary HTTP endpoint. An extension may normalize it to HTTPS. protected String buildRedirectUrlToBackend(HttpServletRequest request, TNetworkAddress addr, String requestPath, String queryString) { - return buildRedirectUrl("http", request, addr, requestPath, queryString); + String url = buildRedirectUrl("http", request, addr, requestPath, queryString); + return InternalHttpClientProviderFactory.getProvider() + .normalizeInternalUrl(url, InternalHttpClientProvider.Target.BE); } private String buildRedirectUrl(String scheme, HttpServletRequest request, TNetworkAddress addr, @@ -146,7 +145,7 @@ public RedirectView redirectTo(HttpServletRequest request, TNetworkAddress addr) return redirectView; } - // Use for redirects whose destination is a BE (e.g. stream load), which never speaks HTTPS. + // Use for redirects whose destination is a BE (for example, stream load). public RedirectView redirectToBackend(HttpServletRequest request, TNetworkAddress addr) { RedirectView redirectView = new RedirectView( buildRedirectUrlToBackend(request, addr, request.getRequestURI(), request.getQueryString())); @@ -318,25 +317,8 @@ public Object forwardToMaster(HttpServletRequest request, @Nullable Object body) HttpEntity entity = new HttpEntity<>(body, headers); - RestTemplate restTemplate; - if (Config.enable_https) { - SimpleClientHttpRequestFactory factory = new SimpleClientHttpRequestFactory() { - @Override - protected void prepareConnection(HttpURLConnection conn, String httpMethod) - throws IOException { - if (conn instanceof HttpsURLConnection) { - HttpsURLConnection https = (HttpsURLConnection) conn; - https.setSSLSocketFactory( - InternalHttpsUtils.getSslContext().getSocketFactory()); - https.setHostnameVerifier(NoopHostnameVerifier.INSTANCE); - } - super.prepareConnection(conn, httpMethod); - } - }; - restTemplate = new RestTemplate(factory); - } else { - restTemplate = new RestTemplate(); - } + RestTemplate restTemplate = InternalHttpClientProviderFactory.getProvider() + .getRestTemplate(InternalHttpClientProvider.Target.FE); ResponseEntity responseEntity; switch (method) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/manager/HttpUtils.java b/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/manager/HttpUtils.java index 43df88ee548506..8c3c70494f3374 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/manager/HttpUtils.java +++ b/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/manager/HttpUtils.java @@ -18,11 +18,13 @@ package org.apache.doris.httpv2.rest.manager; import org.apache.doris.catalog.Env; -import org.apache.doris.common.Config; import org.apache.doris.common.Pair; import org.apache.doris.common.util.HttpURLUtil; import org.apache.doris.common.util.InternalHttpsUtils; +import org.apache.doris.common.util.NetUtils; import org.apache.doris.common.util.Util; +import org.apache.doris.httpv2.client.InternalHttpClientProvider; +import org.apache.doris.httpv2.client.InternalHttpClientProviderFactory; import org.apache.doris.httpv2.entity.ResponseBody; import org.apache.doris.persist.gson.GsonUtils; import org.apache.doris.system.Frontend; @@ -76,8 +78,8 @@ static boolean isCurrentFe(String ip, int port) { } public static String concatUrl(Pair ipPort, String path, Map arguments) { - StringBuilder url = new StringBuilder(Config.enable_https ? "https://" : "http://") - .append(ipPort.first).append(":").append(ipPort.second).append(path); + StringBuilder url = new StringBuilder("http://") + .append(NetUtils.getHostPortInAccessibleFormat(ipPort.first, ipPort.second)).append(path); boolean isFirst = true; for (Map.Entry entry : arguments.entrySet()) { if (!Strings.isNullOrEmpty(entry.getValue())) { @@ -90,7 +92,8 @@ public static String concatUrl(Pair ipPort, String path, Map headers, int timeoutMs) { if (null != headers) { for (String key : headers.keySet()) { @@ -134,16 +162,29 @@ public static CloseableHttpClient getHttpClient() { return HttpClientBuilder.create().build(); } + public static CloseableHttpClient getInternalHttpClient(InternalHttpClientProvider.Target target) { + return InternalHttpClientProviderFactory.getProvider().getHttpClient(target); + } + private static String executeRequest(HttpRequestBase request) throws IOException { - // Pick client by this request's own scheme, since this method also serves plain http BE calls. boolean useHttpsClient = "https".equalsIgnoreCase(request.getURI().getScheme()); try (CloseableHttpClient client = useHttpsClient ? InternalHttpsUtils.createValidatedHttpClient() - : HttpClientBuilder.create().build()) { + : getHttpClient()) { return client.execute(request, httpResponse -> EntityUtils.toString(httpResponse.getEntity())); } } + private static String executeInternalRequest(HttpRequestBase request, + InternalHttpClientProvider.Target target) throws IOException { + CloseableHttpClient client = getInternalHttpClient(target); + return client.execute(request, httpResponse -> EntityUtils.toString(httpResponse.getEntity())); + } + + public static String normalizeInternalUrl(String url, InternalHttpClientProvider.Target target) { + return InternalHttpClientProviderFactory.getProvider().normalizeInternalUrl(url, target); + } + static String parseResponse(String response) { ResponseBody responseEntity = GsonUtils.GSON.fromJson(response, new TypeToken() {}.getType()); if (responseEntity.getCode() != REQUEST_SUCCESS_CODE) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/manager/NodeAction.java b/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/manager/NodeAction.java index 67aa7036f6a178..8f5b199b0dda3e 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/manager/NodeAction.java +++ b/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/manager/NodeAction.java @@ -31,6 +31,7 @@ import org.apache.doris.common.util.NetUtils; import org.apache.doris.common.util.PropertyAnalyzer; import org.apache.doris.ha.FrontendNodeType; +import org.apache.doris.httpv2.client.InternalHttpClientProvider; import org.apache.doris.httpv2.controller.BaseController.ActionAuthorizationInfo; import org.apache.doris.httpv2.entity.ResponseEntityBuilder; import org.apache.doris.httpv2.rest.RestBaseController; @@ -205,7 +206,8 @@ public Object configurationName(HttpServletRequest request, HttpServletResponse Backend be = Env.getCurrentSystemInfo().getBackend(beIds.get(0)); String url = "http://" + NetUtils.getHostPortInAccessibleFormat(be.getHost(), be.getHttpPort()) + "/api/show_config"; - String questResult = HttpUtils.doGet(url, null); + String questResult = HttpUtils.doInternalGet( + url, null, InternalHttpClientProvider.Target.BE); List> configs = GsonUtils.GSON.fromJson(questResult, new TypeToken>>() { }.getType()); for (List config : configs) { @@ -382,13 +384,12 @@ private List> handleConfigurationInfo(List> h Pair hostPort = hostPorts.get(i); String address = NetUtils.getHostPortInAccessibleFormat(hostPort.first, hostPort.second); configRequestDoneSignal.addMark(address, -1); - // FE nodes use HTTPS when enabled; BE nodes always use plain HTTP - // (BEs do not participate in the FE internal HTTPS scheme) - String scheme = (Config.enable_https && "FE".equals(nodeType)) ? "https://" : "http://"; - String url = scheme + address + questPath; + String url = "http://" + address + questPath; + InternalHttpClientProvider.Target target = "FE".equalsIgnoreCase(nodeType) + ? InternalHttpClientProvider.Target.FE : InternalHttpClientProvider.Target.BE; httpExecutor.submit( - new HttpConfigInfoTask(url, hostPort, authorization, nodeType, confNames, configRequestDoneSignal, - configInfoTotal.get(i))); + new HttpConfigInfoTask(url, hostPort, authorization, nodeType, target, confNames, + configRequestDoneSignal, configInfoTotal.get(i))); } List> resultConfigs = Lists.newArrayList(); try { @@ -431,17 +432,19 @@ private class HttpConfigInfoTask implements Runnable { private Pair hostPort; private String authorization; private String nodeType; + private InternalHttpClientProvider.Target target; private List confNames; private MarkedCountDownLatch configRequestDoneSignal; private List> config; public HttpConfigInfoTask(String url, Pair hostPort, String authorization, String nodeType, - List confNames, MarkedCountDownLatch configRequestDoneSignal, - List> config) { + InternalHttpClientProvider.Target target, List confNames, + MarkedCountDownLatch configRequestDoneSignal, List> config) { this.url = url; this.hostPort = hostPort; this.authorization = authorization; this.nodeType = nodeType; + this.target = target; this.confNames = confNames; this.configRequestDoneSignal = configRequestDoneSignal; this.config = config; @@ -451,8 +454,8 @@ public HttpConfigInfoTask(String url, Pair hostPort, String aut public void run() { String configInfo; try { - configInfo = HttpUtils.doGet(url, - ImmutableMap.builder().put(AUTHORIZATION, authorization).build()); + configInfo = HttpUtils.doInternalGet(url, + ImmutableMap.builder().put(AUTHORIZATION, authorization).build(), target); List> configs = GsonUtils.GSON.fromJson(configInfo, new TypeToken>>() { }.getType()); for (List conf : configs) { @@ -520,7 +523,8 @@ public Object setConfigFe(HttpServletRequest request, HttpServletResponse respon if (!nodeConfigs.getConfigs(true).isEmpty()) { String url = concatFeSetConfigUrl(nodeConfigs, true); try { - String responsePersist = HttpUtils.doGet(url, header); + String responsePersist = HttpUtils.doInternalGet( + url, header, InternalHttpClientProvider.Target.FE); parseFeSetConfigResponse(responsePersist, nodeConfigs.getHostPort(), failedTotal); } catch (Exception e) { addSetConfigErrNode(nodeConfigs.getConfigs(true), nodeConfigs.getHostPort(), e.getMessage(), @@ -530,7 +534,8 @@ public Object setConfigFe(HttpServletRequest request, HttpServletResponse respon if (!nodeConfigs.getConfigs(false).isEmpty()) { String url = concatFeSetConfigUrl(nodeConfigs, false); try { - String responseTemp = HttpUtils.doGet(url, header); + String responseTemp = HttpUtils.doInternalGet( + url, header, InternalHttpClientProvider.Target.FE); parseFeSetConfigResponse(responseTemp, nodeConfigs.getHostPort(), failedTotal); } catch (Exception e) { addSetConfigErrNode(nodeConfigs.getConfigs(false), nodeConfigs.getHostPort(), e.getMessage(), @@ -582,8 +587,9 @@ private static void addFailedConfig(String configName, String value, String node private String concatFeSetConfigUrl(NodeConfigs nodeConfigs, boolean isPersist) { StringBuilder sb = new StringBuilder(); Pair hostPort = nodeConfigs.getHostPort(); - sb.append(Config.enable_https ? "https://" : "http://") - .append(hostPort.first).append(":").append(hostPort.second).append("/api/_set_config"); + sb.append("http://") + .append(NetUtils.getHostPortInAccessibleFormat(hostPort.first, hostPort.second)) + .append("/api/_set_config"); Map configs = nodeConfigs.getConfigs(isPersist); boolean addAnd = false; for (Map.Entry entry : configs.entrySet()) { @@ -881,7 +887,8 @@ private void submitBeSetConfigTask(NodeConfigs nodeConfigs, boolean isPersist, S private String concatBeSetConfigUrl(String host, Integer port, String configName, String configValue, boolean isPersist) { StringBuilder stringBuffer = new StringBuilder(); - stringBuffer.append("http://").append(host).append(":").append(port).append("/api/update_config").append("?") + stringBuffer.append("http://").append(NetUtils.getHostPortInAccessibleFormat(host, port)) + .append("/api/update_config").append("?") .append(configName).append("=").append(configValue); if (isPersist) { stringBuffer.append("&persist=true"); @@ -927,8 +934,9 @@ public HttpSetConfigTask(String url, Pair hostPort, String auth @Override public void run() { try { - String response = HttpUtils.doPost(url, - ImmutableMap.builder().put(AUTHORIZATION, authorization).build(), null); + String response = HttpUtils.doInternalPost(url, + ImmutableMap.builder().put(AUTHORIZATION, authorization).build(), null, + InternalHttpClientProvider.Target.BE); JsonObject jsonObject = JsonParser.parseString(response).getAsJsonObject(); String status = jsonObject.get("status").getAsString(); if (!status.equals("OK")) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/manager/QueryProfileAction.java b/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/manager/QueryProfileAction.java index d0aae7d1a0ebda..54e48cfce25ca8 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/manager/QueryProfileAction.java +++ b/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/manager/QueryProfileAction.java @@ -29,6 +29,7 @@ import org.apache.doris.common.profile.SummaryProfile; import org.apache.doris.common.util.HttpURLUtil; import org.apache.doris.common.util.NetUtils; +import org.apache.doris.httpv2.client.InternalHttpClientProvider; import org.apache.doris.httpv2.controller.BaseController.ActionAuthorizationInfo; import org.apache.doris.httpv2.entity.ResponseEntityBuilder; import org.apache.doris.httpv2.rest.RestBaseController; @@ -134,9 +135,11 @@ private List requestAllFe(String httpPath, Map arguments try { String data = null; if (method == HttpMethod.GET) { - data = HttpUtils.parseResponse(HttpUtils.doGet(url, header)); + data = HttpUtils.parseResponse(HttpUtils.doInternalGet( + url, header, InternalHttpClientProvider.Target.FE)); } else if (method == HttpMethod.POST) { - data = HttpUtils.parseResponse(HttpUtils.doPost(url, header, null)); + data = HttpUtils.parseResponse(HttpUtils.doInternalPost( + url, header, null, InternalHttpClientProvider.Target.FE)); } if (!Strings.isNullOrEmpty(data) && !data.equals("{}")) { dataList.add(data); @@ -349,7 +352,8 @@ private String getQueryIdByTraceIdImpl(HttpServletRequest request, String traceI continue; } String url = HttpUtils.concatUrl(ipPort, httpPath, arguments); - String responseJson = HttpUtils.doGet(url, header); + String responseJson = HttpUtils.doInternalGet( + url, header, InternalHttpClientProvider.Target.FE); JsonObject jObj = JsonParser.parseString(responseJson).getAsJsonObject(); int code = jObj.get("code").getAsInt(); if (code == HttpUtils.REQUEST_SUCCESS_CODE) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/httpv2/restv2/ClusterGuardAction.java b/fe/fe-core/src/main/java/org/apache/doris/httpv2/restv2/ClusterGuardAction.java index a41509fd0a87a6..95b75539038f05 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/httpv2/restv2/ClusterGuardAction.java +++ b/fe/fe-core/src/main/java/org/apache/doris/httpv2/restv2/ClusterGuardAction.java @@ -21,6 +21,7 @@ import org.apache.doris.cluster.ClusterGuardException; import org.apache.doris.cluster.ClusterGuardFactory; import org.apache.doris.common.Pair; +import org.apache.doris.httpv2.client.InternalHttpClientProvider; import org.apache.doris.httpv2.entity.ResponseEntityBuilder; import org.apache.doris.httpv2.rest.RestBaseController; import org.apache.doris.httpv2.rest.manager.HttpUtils; @@ -124,7 +125,8 @@ private Object reloadOnAllFe(HttpServletRequest request) { String nodeKey = ipPort.first + ":" + ipPort.second; String url = HttpUtils.concatUrl(ipPort, httpPath, arguments); try { - String resp = HttpUtils.doPost(url, header, null); + String resp = HttpUtils.doInternalPost( + url, header, null, InternalHttpClientProvider.Target.FE); JsonObject jsonObj = JsonParser.parseString(resp).getAsJsonObject(); int code = jsonObj.get("code").getAsInt(); if (code == HttpUtils.REQUEST_SUCCESS_CODE) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/MysqlLoadManager.java b/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/MysqlLoadManager.java index c812462fae86d1..9e910ff96c4810 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/MysqlLoadManager.java +++ b/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/MysqlLoadManager.java @@ -31,8 +31,10 @@ import org.apache.doris.common.ThreadPoolManager; import org.apache.doris.common.UserException; import org.apache.doris.common.io.ByteBufferNetworkInputStream; +import org.apache.doris.common.util.NetUtils; import org.apache.doris.datasource.property.fileformat.CsvFileFormatProperties; import org.apache.doris.datasource.property.fileformat.FileFormatProperties; +import org.apache.doris.httpv2.client.InternalHttpClientProvider; import org.apache.doris.httpv2.rest.manager.HttpUtils; import org.apache.doris.load.LoadJobRowResult; import org.apache.doris.load.StreamLoadHandler; @@ -62,7 +64,6 @@ import org.apache.http.entity.ContentType; import org.apache.http.entity.InputStreamEntity; import org.apache.http.impl.client.CloseableHttpClient; -import org.apache.http.impl.client.HttpClients; import org.apache.http.util.EntityUtils; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; @@ -188,7 +189,9 @@ public LoadJobRowResult executeMySqlLoadJob(ConnectContext context, MysqlDataDes MySqlLoadContext loadContext = new MySqlLoadContext(); loadContextMap.put(loadId, loadContext); LOG.info("Executing mysql load with id: {}.", loadId); - try (final CloseableHttpClient httpclient = HttpUtils.getHttpClient()) { + final CloseableHttpClient httpclient = + HttpUtils.getInternalHttpClient(InternalHttpClientProvider.Target.BE); + try { for (String file : filePaths) { InputStreamEntity entity = getInputStreamEntity(context, clientLocal, file, loadId); HttpPut request = generateRequestForMySqlLoadV2(context, entity, dataDesc, database, table, token); @@ -255,7 +258,9 @@ public LoadJobRowResult executeMySqlLoadJobFromCommand(ConnectContext context, N MySqlLoadContext loadContext = new MySqlLoadContext(); loadContextMap.put(loadId, loadContext); LOG.info("Executing mysql load with id: {}.", loadId); - try (final CloseableHttpClient httpclient = HttpClients.createDefault()) { + final CloseableHttpClient httpclient = + HttpUtils.getInternalHttpClient(InternalHttpClientProvider.Target.BE); + try { for (String file : filePaths) { InputStreamEntity entity = getInputStreamEntity(context, clientLocal, file, loadId); HttpPut request = generateRequestForMySqlLoad(context, entity, dataDesc, database, table, token); @@ -706,14 +711,12 @@ private String selectBackendForMySqlLoad(ConnectContext context, String database StringBuilder sb = new StringBuilder(); sb.append("http://"); - sb.append(backend.getHost()); - sb.append(":"); - sb.append(backend.getHttpPort()); + sb.append(NetUtils.getHostPortInAccessibleFormat(backend.getHost(), backend.getHttpPort())); sb.append("/api/"); sb.append(database); sb.append("/"); sb.append(table); sb.append("/_stream_load"); - return sb.toString(); + return HttpUtils.normalizeInternalUrl(sb.toString(), InternalHttpClientProvider.Target.BE); } } diff --git a/fe/fe-core/src/main/java/org/apache/doris/master/MetaHelper.java b/fe/fe-core/src/main/java/org/apache/doris/master/MetaHelper.java index 73fceec8b39481..37c14d6bb6420e 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/master/MetaHelper.java +++ b/fe/fe-core/src/main/java/org/apache/doris/master/MetaHelper.java @@ -21,6 +21,7 @@ import org.apache.doris.common.Config; import org.apache.doris.common.io.IOUtils; import org.apache.doris.common.util.HttpURLUtil; +import org.apache.doris.httpv2.client.InternalHttpClientProvider; import org.apache.doris.httpv2.entity.ResponseBody; import org.apache.doris.httpv2.rest.manager.HttpUtils; import org.apache.doris.persist.gson.GsonUtils; @@ -150,7 +151,8 @@ private static void checkFile(File file) throws IOException { public static ResponseBody doGet(String url, int timeout, Class clazz) throws IOException { Map headers = HttpURLUtil.getNodeIdentHeaders(); LOG.info("meta helper, url: {}, timeout: {}, header names: {}", url, timeout, headers.keySet()); - String response = HttpUtils.doGet(url, headers, timeout); + String response = HttpUtils.doInternalGet( + url, headers, timeout, InternalHttpClientProvider.Target.FE); try { return parseResponse(response, clazz); } catch (Exception e) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/minidump/MinidumpUtils.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/minidump/MinidumpUtils.java index 43bde3f23e1162..14e45cf1554115 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/minidump/MinidumpUtils.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/minidump/MinidumpUtils.java @@ -28,7 +28,9 @@ import org.apache.doris.common.Config; import org.apache.doris.common.proc.FrontendsProcNode; import org.apache.doris.common.util.DebugUtil; -import org.apache.doris.common.util.HttpURLUtil; +import org.apache.doris.common.util.NetUtils; +import org.apache.doris.httpv2.client.InternalHttpClientProvider; +import org.apache.doris.httpv2.client.InternalHttpClientProviderFactory; import org.apache.doris.nereids.NereidsPlanner; import org.apache.doris.nereids.StatementContext; import org.apache.doris.nereids.glue.LogicalPlanAdapter; @@ -99,17 +101,22 @@ public static String getHttpGetString() { return HTTP_GET_STRING; } + static String buildHttpGetString(String feAddress, int feHttpPort, String queryId) { + String url = "http://" + NetUtils.getHostPortInAccessibleFormat(feAddress, feHttpPort) + + "/api/minidump?query_id=" + queryId; + return InternalHttpClientProviderFactory.getProvider().normalizeInternalUrl( + url, InternalHttpClientProvider.Target.FE); + } + /** * Saving of minidump file to fe log path */ public static void saveMinidumpString(JSONObject minidump, String querId) { String dumpPath = MinidumpUtils.DUMP_PATH + File.separator + "_" + querId; String feAddress = FrontendsProcNode.getCurrentFrontendVersion(Env.getCurrentEnv()).getHost(); - int feHttpPort = HttpURLUtil.getHttpPort(); - String scheme = Config.enable_https ? "https" : "http"; + int feHttpPort = Config.http_port; MinidumpUtils.DUMP_FILE_FULL_PATH = dumpPath + ".json"; - MinidumpUtils.HTTP_GET_STRING = scheme + "://" + feAddress + ":" + feHttpPort - + "/api/minidump?query_id=" + querId; + MinidumpUtils.HTTP_GET_STRING = buildHttpGetString(feAddress, feHttpPort, querId); String jsonMinidump = minidump.toString(4); try (FileWriter file = new FileWriter(MinidumpUtils.DUMP_FILE_FULL_PATH)) { file.write(jsonMinidump); diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/ShowConfigCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/ShowConfigCommand.java index d106a63897fbf6..71bb7c36f29ee9 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/ShowConfigCommand.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/ShowConfigCommand.java @@ -28,6 +28,9 @@ import org.apache.doris.common.ErrorReport; import org.apache.doris.common.PatternMatcher; import org.apache.doris.common.PatternMatcherWrapper; +import org.apache.doris.common.util.HttpURLUtil; +import org.apache.doris.common.util.NetUtils; +import org.apache.doris.httpv2.client.InternalHttpClientProvider; import org.apache.doris.mysql.privilege.PrivPredicate; import org.apache.doris.nereids.trees.plans.PlanType; import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor; @@ -46,8 +49,7 @@ import java.io.BufferedReader; import java.io.InputStream; import java.io.InputStreamReader; -import java.net.URL; -import java.net.URLConnection; +import java.net.HttpURLConnection; import java.util.ArrayList; import java.util.Comparator; import java.util.List; @@ -126,10 +128,11 @@ private ShowResultSet handShowBackendConfig() throws AnalysisException { Backend backend = systemInfoService.getBackend(beId); String host = backend.getHost(); int httpPort = backend.getHttpPort(); - String urlString = String.format("http://%s:%d/api/show_config", host, httpPort); + String urlString = String.format("http://%s/api/show_config", + NetUtils.getHostPortInAccessibleFormat(host, httpPort)); try { - URL url = new URL(urlString); - URLConnection urlConnection = url.openConnection(); + HttpURLConnection urlConnection = HttpURLUtil.getInternalConnection( + urlString, InternalHttpClientProvider.Target.BE); urlConnection.setRequestProperty("Auth-Token", Env.getCurrentEnv().getTokenManager().acquireToken()); InputStream inputStream = urlConnection.getInputStream(); BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream)); diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/ShowLoadWarningsCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/ShowLoadWarningsCommand.java index cfd307534a8dcf..c85ab1f4641411 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/ShowLoadWarningsCommand.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/ShowLoadWarningsCommand.java @@ -29,7 +29,9 @@ import org.apache.doris.common.ErrorCode; import org.apache.doris.common.ErrorReport; import org.apache.doris.common.FeConstants; +import org.apache.doris.common.util.HttpURLUtil; import org.apache.doris.common.util.NetUtils; +import org.apache.doris.httpv2.client.InternalHttpClientProvider; import org.apache.doris.load.loadv2.LoadManager; import org.apache.doris.nereids.analyzer.UnboundSlot; import org.apache.doris.nereids.trees.expressions.And; @@ -58,9 +60,9 @@ import java.io.BufferedReader; import java.io.InputStream; import java.io.InputStreamReader; +import java.net.HttpURLConnection; import java.net.MalformedURLException; import java.net.URL; -import java.net.URLConnection; import java.util.Collections; import java.util.List; import java.util.Objects; @@ -234,7 +236,8 @@ private ShowResultSet handleShowLoadWarningsFromURL(URL url) throws AnalysisExce List> rows = Lists.newArrayList(); try { - URLConnection urlConnection = url.openConnection(); + HttpURLConnection urlConnection = HttpURLUtil.getInternalConnection( + url.toString(), InternalHttpClientProvider.Target.BE); urlConnection.setRequestProperty("Auth-Token", Env.getCurrentEnv().getTokenManager().acquireToken()); InputStream inputStream = urlConnection.getInputStream(); try (BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream))) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/plugin/audit/AuditStreamLoader.java b/fe/fe-core/src/main/java/org/apache/doris/plugin/audit/AuditStreamLoader.java index 73a78e7356ed66..3c34049a5a9213 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/plugin/audit/AuditStreamLoader.java +++ b/fe/fe-core/src/main/java/org/apache/doris/plugin/audit/AuditStreamLoader.java @@ -22,10 +22,9 @@ import org.apache.doris.common.Config; import org.apache.doris.common.FeConstants; import org.apache.doris.common.util.HttpURLUtil; -import org.apache.doris.common.util.InternalHttpsUtils; +import org.apache.doris.httpv2.client.InternalHttpClientProvider; import org.apache.doris.qe.GlobalVariable; -import org.apache.http.conn.ssl.NoopHostnameVerifier; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; @@ -34,10 +33,8 @@ import java.io.IOException; import java.io.InputStreamReader; import java.net.HttpURLConnection; -import java.net.URL; import java.util.Calendar; import java.util.stream.Collectors; -import javax.net.ssl.HttpsURLConnection; public class AuditStreamLoader { private static final Logger LOG = LogManager.getLogger(AuditStreamLoader.class); @@ -51,21 +48,25 @@ public class AuditStreamLoader { public AuditStreamLoader() { this.db = FeConstants.INTERNAL_DB_NAME; this.auditLogTbl = AuditLoader.AUDIT_LOG_TABLE; - String scheme = Config.enable_https ? "https" : "http"; - String hostPort = "127.0.0.1:" + HttpURLUtil.getHttpPort(); - this.auditLogLoadUrlStr = scheme + "://" + hostPort + "/api/" + db + "/" + auditLogTbl + "/_stream_load?"; + String hostPort = "127.0.0.1:" + Config.http_port; + this.auditLogLoadUrlStr = "http://" + hostPort + "/api/" + db + "/" + auditLogTbl + "/_stream_load?"; // currently, FE identity is FE's IP:port, so we replace the "." and ":" to make it suitable for label this.feIdentity = Env.getCurrentEnv().getSelfNode().getIdent().replaceAll("\\.", "_").replaceAll(":", "_"); } - private HttpURLConnection getConnection(String urlStr, String label, String clusterToken) throws IOException { - URL url = new URL(urlStr); - HttpURLConnection conn = (HttpURLConnection) url.openConnection(); - if (conn instanceof HttpsURLConnection && Config.enable_https) { - HttpsURLConnection httpsConn = (HttpsURLConnection) conn; - httpsConn.setSSLSocketFactory(InternalHttpsUtils.getSslContext().getSocketFactory()); - httpsConn.setHostnameVerifier(NoopHostnameVerifier.INSTANCE); - } + private HttpURLConnection getFeConnection(String urlStr, String label, String clusterToken) throws IOException { + HttpURLConnection conn = getConnection(urlStr, label, clusterToken, InternalHttpClientProvider.Target.FE); + conn.addRequestProperty("redirect-policy", "random-be"); + return conn; + } + + private HttpURLConnection getBeConnection(String urlStr, String label, String clusterToken) throws IOException { + return getConnection(urlStr, label, clusterToken, InternalHttpClientProvider.Target.BE); + } + + private HttpURLConnection getConnection(String urlStr, String label, String clusterToken, + InternalHttpClientProvider.Target target) throws IOException { + HttpURLConnection conn = HttpURLUtil.getInternalConnection(urlStr, target); conn.setInstanceFollowRedirects(false); conn.setRequestMethod("PUT"); conn.setRequestProperty("token", clusterToken); @@ -80,7 +81,6 @@ private HttpURLConnection getConnection(String urlStr, String label, String clus conn.addRequestProperty("columns", InternalSchema.AUDIT_SCHEMA.stream().map(c -> c.getName()).collect( Collectors.joining(","))); - conn.addRequestProperty("redirect-policy", "random-be"); conn.addRequestProperty("column_separator", AuditLoader.AUDIT_TABLE_COL_SEPARATOR_STR); conn.addRequestProperty("line_delimiter", AuditLoader.AUDIT_TABLE_LINE_DELIMITER_STR); conn.addRequestProperty("skip_record_to_audit_log_table", "true"); @@ -132,7 +132,7 @@ public LoadResponse loadBatch(StringBuilder sb, String clusterToken) { try { // build request and send to fe label = "audit" + label; - feConn = getConnection(auditLogLoadUrlStr, label, clusterToken); + feConn = getFeConnection(auditLogLoadUrlStr, label, clusterToken); int status = feConn.getResponseCode(); // fe send back http response code TEMPORARY_REDIRECT 307 and new be location if (status != 307) { @@ -144,7 +144,7 @@ public LoadResponse loadBatch(StringBuilder sb, String clusterToken) { throw new Exception("redirect location is null"); } // build request and send to new be location - beConn = getConnection(location, label, clusterToken); + beConn = getBeConnection(location, label, clusterToken); // send data to be try (BufferedOutputStream bos = new BufferedOutputStream(beConn.getOutputStream())) { bos.write(sb.toString().getBytes()); diff --git a/fe/fe-core/src/main/java/org/apache/doris/tls/server/TlsProtocolSet.java b/fe/fe-core/src/main/java/org/apache/doris/tls/server/TlsProtocolSet.java index 6745c6ec481f0c..3373c0c2923dec 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/tls/server/TlsProtocolSet.java +++ b/fe/fe-core/src/main/java/org/apache/doris/tls/server/TlsProtocolSet.java @@ -45,6 +45,10 @@ public static synchronized boolean isProtocolIncluded(Protocol protocol) { return !excludedProtocols().contains(protocol.name().toLowerCase(Locale.ROOT)); } + public static boolean isHttpTlsActive() { + return Config.enable_tls && isProtocolIncluded(Protocol.HTTP); + } + private static Set excludedProtocols() { String raw = Config.tls_excluded_protocols == null ? "" : Config.tls_excluded_protocols; if (raw.equals(cachedRaw)) { diff --git a/fe/fe-core/src/test/java/org/apache/doris/httpv2/client/OssInternalHttpClientProviderTest.java b/fe/fe-core/src/test/java/org/apache/doris/httpv2/client/OssInternalHttpClientProviderTest.java new file mode 100644 index 00000000000000..4b35dd12b23bfa --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/httpv2/client/OssInternalHttpClientProviderTest.java @@ -0,0 +1,120 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package org.apache.doris.httpv2.client; + +import org.apache.doris.common.Config; + +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +public class OssInternalHttpClientProviderTest { + private boolean oldEnableHttps; + private int oldHttpsPort; + private boolean oldEnableTls; + private String oldTlsExcludedProtocols; + + @Before + public void setUp() { + oldEnableHttps = Config.enable_https; + oldHttpsPort = Config.https_port; + oldEnableTls = Config.enable_tls; + oldTlsExcludedProtocols = Config.tls_excluded_protocols; + Config.enable_https = false; + Config.https_port = 8443; + Config.enable_tls = false; + Config.tls_excluded_protocols = ""; + } + + @After + public void tearDown() { + Config.enable_https = oldEnableHttps; + Config.https_port = oldHttpsPort; + Config.enable_tls = oldEnableTls; + Config.tls_excluded_protocols = oldTlsExcludedProtocols; + } + + @Test + public void testDefaultConfigurationDoesNotRewriteUrl() { + OssInternalHttpClientProvider provider = new OssInternalHttpClientProvider(); + + Assert.assertEquals("http://fe-host:8080/rest/v1/session", + provider.normalizeInternalUrl("http://fe-host:8080/rest/v1/session", + InternalHttpClientProvider.Target.FE)); + } + + @Test + public void testFeUrlRewrittenWhenOpenSourceHttpsEnabled() { + Config.enable_https = true; + OssInternalHttpClientProvider provider = new OssInternalHttpClientProvider(); + + Assert.assertEquals("https://fe-host:8443/rest/v1/session", + provider.normalizeInternalUrl("http://fe-host:8080/rest/v1/session", + InternalHttpClientProvider.Target.FE)); + } + + @Test + public void testEncodedUrlComponentsArePreserved() { + Config.enable_https = true; + OssInternalHttpClientProvider provider = new OssInternalHttpClientProvider(); + + Assert.assertEquals("https://fe-host:8443/rest/v1/query?search=a%2Bb#part%201", + provider.normalizeInternalUrl("http://fe-host:8080/rest/v1/query?search=a%2Bb#part%201", + InternalHttpClientProvider.Target.FE)); + } + + @Test + public void testHttpsInputIsIdempotent() { + Config.enable_https = true; + OssInternalHttpClientProvider provider = new OssInternalHttpClientProvider(); + + Assert.assertEquals("https://fe-host:9443/rest/v1/session", + provider.normalizeInternalUrl("https://fe-host:9443/rest/v1/session", + InternalHttpClientProvider.Target.FE)); + } + + @Test + public void testBeUrlIsNotRewrittenByOpenSourceHttps() { + Config.enable_https = true; + OssInternalHttpClientProvider provider = new OssInternalHttpClientProvider(); + + Assert.assertEquals("http://be-host:8040/api/show_config", + provider.normalizeInternalUrl("http://be-host:8040/api/show_config", + InternalHttpClientProvider.Target.BE)); + } + + @Test(expected = UnsupportedOperationException.class) + public void testHttpTlsRequiresCustomProvider() { + Config.enable_tls = true; + OssInternalHttpClientProvider provider = new OssInternalHttpClientProvider(); + + provider.normalizeInternalUrl("http://fe-host:8030/check", InternalHttpClientProvider.Target.FE); + } + + @Test + public void testExcludedHttpTlsKeepsOssProviderAvailable() { + Config.enable_tls = true; + Config.tls_excluded_protocols = "brpc, HTTP "; + OssInternalHttpClientProvider provider = new OssInternalHttpClientProvider(); + + Assert.assertEquals("http://fe-host:8030/check", + provider.normalizeInternalUrl( + "http://fe-host:8030/check", InternalHttpClientProvider.Target.FE)); + } +} diff --git a/fe/fe-core/src/test/java/org/apache/doris/httpv2/rest/LoadActionTest.java b/fe/fe-core/src/test/java/org/apache/doris/httpv2/rest/LoadActionTest.java index 93bd89fafd1844..860e652c1ba1eb 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/httpv2/rest/LoadActionTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/httpv2/rest/LoadActionTest.java @@ -51,6 +51,8 @@ public class LoadActionTest { private final String originalCloudUniqueId = Config.cloud_unique_id; private final boolean originalEnableGroupCommitStreamLoadBeForward = Config.enable_group_commit_streamload_be_forward; + private final boolean originalEnableTls = Config.enable_tls; + private final String originalTlsExcludedProtocols = Config.tls_excluded_protocols; @AfterEach @@ -59,6 +61,8 @@ public void tearDown() { Config.enable_debug_points = originalEnableDebugPoints; Config.cloud_unique_id = originalCloudUniqueId; Config.enable_group_commit_streamload_be_forward = originalEnableGroupCommitStreamLoadBeForward; + Config.enable_tls = originalEnableTls; + Config.tls_excluded_protocols = originalTlsExcludedProtocols; DebugPointUtil.clearDebugPoints(); Thread.interrupted(); org.apache.doris.qe.ConnectContext.remove(); @@ -392,6 +396,34 @@ public void testRedirectToStreamLoadForwardBuildsForwardUrl() throws Exception { redirectView.getUrl()); } + @Test + public void testHttpTlsSkipsGroupCommitBeForward() throws Exception { + Config.enable_debug_points = true; + Config.cloud_unique_id = "cloud-mode"; + Config.enable_group_commit_streamload_be_forward = true; + Config.enable_tls = true; + Config.tls_excluded_protocols = ""; + DebugPointUtil.addDebugPointWithValue("LoadAction.selectRedirectBackend.backendId", 1L); + + LoadAction loadAction = new LoadAction(); + HttpServletRequest request = Mockito.mock(HttpServletRequest.class); + Backend backend = mockBackend("group-commit-be", 8040, null); + SystemInfoService systemInfoService = Mockito.mock(SystemInfoService.class); + Mockito.when(systemInfoService.getBackend(1L)).thenReturn(backend); + + try (MockedStatic mockedEnv = Mockito.mockStatic(Env.class); + MockedStatic mockedStreamLoad = Mockito.mockStatic(StreamLoadHandler.class)) { + mockedEnv.when(Env::getCurrentSystemInfo).thenReturn(systemInfoService); + + TNetworkAddress addr = invokeHandleStreamLoadRedirect( + loadAction, request, true, 10L, "db1", "tbl1", "label1"); + + Assertions.assertEquals("group-commit-be", addr.getHostname()); + Assertions.assertEquals(8040, addr.getPort()); + mockedStreamLoad.verifyNoInteractions(); + } + } + @Test public void testSelectCloudRedirectBackendIgnoresLoadSelection() throws Exception { LoadAction loadAction = new LoadAction(); @@ -523,6 +555,14 @@ private RedirectView invokeRedirectToStreamLoadForward(LoadAction loadAction, Ht return (RedirectView) method.invoke(loadAction, request, addr, forwardTarget); } + private TNetworkAddress invokeHandleStreamLoadRedirect(LoadAction loadAction, HttpServletRequest request, + boolean groupCommit, long tableId, String dbName, String tableName, String label) throws Exception { + Method method = LoadAction.class.getDeclaredMethod("handleStreamLoadRedirect", + HttpServletRequest.class, boolean.class, long.class, String.class, String.class, String.class); + method.setAccessible(true); + return (TNetworkAddress) method.invoke(loadAction, request, groupCommit, tableId, dbName, tableName, label); + } + private String invokeGetCloudClusterName(LoadAction loadAction, HttpServletRequest request) throws Exception { Method method = LoadAction.class.getDeclaredMethod("getCloudClusterName", HttpServletRequest.class); method.setAccessible(true); diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/minidump/MinidumpUtTest.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/minidump/MinidumpUtTest.java index 7c01338955c08f..7556fa1de9717b 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/nereids/minidump/MinidumpUtTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/minidump/MinidumpUtTest.java @@ -20,6 +20,8 @@ import org.apache.doris.catalog.Env; import org.apache.doris.common.Config; import org.apache.doris.common.proc.FrontendsProcNode; +import org.apache.doris.httpv2.client.InternalHttpClientProvider; +import org.apache.doris.httpv2.client.InternalHttpClientProviderFactory; import org.apache.doris.system.Frontend; import org.json.JSONObject; @@ -37,6 +39,40 @@ */ class MinidumpUtTest { + @Test + public void testHttpGetStringUsesInternalHttpProvider() { + InternalHttpClientProvider provider = Mockito.mock(InternalHttpClientProvider.class); + Mockito.when(provider.normalizeInternalUrl( + "http://fe-host:8030/api/minidump?query_id=query-1", + InternalHttpClientProvider.Target.FE)) + .thenReturn("https://fe-host:8030/api/minidump?query_id=query-1"); + + try (MockedStatic factory = + Mockito.mockStatic(InternalHttpClientProviderFactory.class)) { + factory.when(InternalHttpClientProviderFactory::getProvider).thenReturn(provider); + + Assertions.assertEquals("https://fe-host:8030/api/minidump?query_id=query-1", + MinidumpUtils.buildHttpGetString("fe-host", 8030, "query-1")); + } + } + + @Test + public void testHttpGetStringFormatsIpv6Address() { + InternalHttpClientProvider provider = Mockito.mock(InternalHttpClientProvider.class); + Mockito.when(provider.normalizeInternalUrl( + "http://[2001:db8::1]:8030/api/minidump?query_id=query-2", + InternalHttpClientProvider.Target.FE)) + .thenAnswer(invocation -> invocation.getArgument(0)); + + try (MockedStatic factory = + Mockito.mockStatic(InternalHttpClientProviderFactory.class)) { + factory.when(InternalHttpClientProviderFactory::getProvider).thenReturn(provider); + + Assertions.assertEquals("http://[2001:db8::1]:8030/api/minidump?query_id=query-2", + MinidumpUtils.buildHttpGetString("2001:db8::1", 8030, "query-2")); + } + } + @Disabled @Test public void testMinidumpUt() {