[feature](fe) Add internal HTTP client provider SPI

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.
@@ -83,15 +76,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);
     }
 
 }

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.
+ *
+ * <p>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);
+}

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<InternalHttpClientProvider> loader = ServiceLoader.load(InternalHttpClientProvider.class);
+        Iterator<InternalHttpClientProvider> 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();
+    }
+}

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;
+        }
+    }
+}

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,15 +38,13 @@
 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;
 import org.springframework.http.HttpHeaders;
 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 {
 
@@ -318,25 +315,8 @@ public Object forwardToMaster(HttpServletRequest request, @Nullable Object body)
 
             HttpEntity<Object> 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<Object> responseEntity;
             switch (method) {