You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@ozone.apache.org by "hevinhsu (via GitHub)" <gi...@apache.org> on 2025/07/03 09:24:57 UTC
[PR] HDDS-11679. Support multiple S3Gs in MiniOzoneCluster [ozone]
hevinhsu opened a new pull request, #8733:
URL: https://github.com/apache/ozone/pull/8733
## What changes were proposed in this pull request?
* Implement a HTTP proxy to forward requests to multiple s3 gateways.
* Implement ```MiniOzoneCluster.Service``` interface to support proxy server and multiple s3 gateway creation.
## What is the link to the Apache JIRA
https://issues.apache.org/jira/browse/HDDS-11679
## How was this patch tested?
* workflow run on fork git repo:
https://github.com/hevinhsu/ozone/actions/runs/16042975855
* workflow runs with ```MultipleS3GatewayService``` to run ```integration-test-s3g``` test cases in another branch on personal repo
https://github.com/hevinhsu/ozone/actions/runs/16043512252
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-11679. Support multiple S3Gs in MiniOzoneCluster [ozone]
Posted by "ivandika3 (via GitHub)" <gi...@apache.org>.
ivandika3 commented on code in PR #8733:
URL: https://github.com/apache/ozone/pull/8733#discussion_r2184473765
##########
hadoop-ozone/integration-test-s3/src/test/java/org/apache/hadoop/ozone/s3/ProxyServer.java:
##########
@@ -0,0 +1,251 @@
+/*
+ * 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.hadoop.ozone.s3;
+
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpHandler;
+import com.sun.net.httpserver.HttpServer;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.HttpURLConnection;
+import java.net.InetSocketAddress;
+import java.net.URL;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import org.apache.commons.io.IOUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * ProxyServer acts as a reverse proxy for S3G endpoints.
+ * It forwards requests to the selected S3G endpoint based on the load balancing strategy.
+ */
+public class ProxyServer {
+ private static final Logger LOG = LoggerFactory.getLogger(ProxyServer.class);
+ private final List<String> s3gEndpoints;
+ private final LoadBalanceStrategy loadBalanceStrategy;
+ private final HttpServer server;
+ private final String host;
+ private final int port;
+ private final ExecutorService executor;
+
+ // Add timeout configurations for better handling of large file operations
+ private static final int CONNECT_TIMEOUT_MS = 60000; // 60 seconds
+ private static final int READ_TIMEOUT_MS = 300000; // 5 minutes
+
+ private static final int BUFFER_SIZE = 64 * 1024; // 64KB
+ private static final ThreadLocal<byte[]> IO_BUFFER = ThreadLocal.withInitial(() -> new byte[BUFFER_SIZE]);
+
+ public ProxyServer(List<String> s3gEndpoints, String host, int proxyPort) throws Exception {
+ this(s3gEndpoints, host, proxyPort, new RoundRobinStrategy());
+ }
+
+ public ProxyServer(List<String> s3gEndpoints, String host, int proxyPort,
+ LoadBalanceStrategy loadBalanceStrategy) throws Exception {
+ this.s3gEndpoints = s3gEndpoints;
+ this.loadBalanceStrategy = loadBalanceStrategy;
+ this.host = host;
+ this.port = proxyPort;
+ server = HttpServer.create(new InetSocketAddress(host, proxyPort), 0);
+ server.createContext("/", new ProxyHandler());
+
+ this.executor = Executors.newCachedThreadPool();
+ server.setExecutor(executor);
+
+ LOG.info("ProxyServer initialized with endpoints: {}", s3gEndpoints);
+ LOG.info("Load balance strategy: {}", loadBalanceStrategy.getClass().getSimpleName());
+ LOG.info("Timeout settings - Connect: {}ms, Read: {}ms", CONNECT_TIMEOUT_MS, READ_TIMEOUT_MS);
+ }
+
+ public void start() {
+ server.start();
+ LOG.info("Proxy started on http://{}:{}", host, port);
+
+ }
+
+ public void stop() {
+ server.stop(0);
+ executor.shutdownNow();
+ LOG.info("Proxy stopped on http://{}:{}", host, port);
+ }
+
+ private class ProxyHandler implements HttpHandler {
+ @Override
+ public void handle(HttpExchange exchange) throws IOException {
+ HttpURLConnection conn = null;
+ try {
+ String target = loadBalanceStrategy.selectEndpoint(s3gEndpoints) + exchange.getRequestURI().toString();
+ conn = createConnection(exchange, target);
+ String requestMethod = conn.getRequestMethod();
+
+ LOG.info("Received request and forwarding request to [{}] {}", requestMethod, target);
+
+ copyRequestHeaders(exchange, conn);
+ addForwardedHeaders(exchange, conn);
+ copyRequestBody(exchange, conn);
+
+ int responseCode = conn.getResponseCode();
+ copyResponseHeaders(exchange, conn);
+
+ if (requestMethod.equals("HEAD") || responseCode == HttpURLConnection.HTTP_NO_CONTENT) {
+ exchange.sendResponseHeaders(responseCode, 0);
+ return;
+ }
+
+ copyResponseBody(exchange, conn, responseCode);
+
+ } catch (Exception e) {
+ LOG.error("Error forwarding request to S3G endpoint", e);
+ sendErrorResponse(exchange, e);
+ } finally {
+ closeResources(conn, exchange);
+ }
+ }
+
+ private void sendErrorResponse(HttpExchange exchange, Exception e) {
+ try {
+ exchange.sendResponseHeaders(502, 0);
+ try (OutputStream os = exchange.getResponseBody()) {
+ os.write(("Proxy error: " + e.getMessage()).getBytes(StandardCharsets.UTF_8));
+ }
+ } catch (IOException ioException) {
+ LOG.error("Failed to send error response", ioException);
+ }
+ }
+
+ private void closeResources(HttpURLConnection conn, HttpExchange exchange) {
+ if (conn != null) {
+ conn.disconnect();
+ }
+ exchange.close();
+ }
+
+ private void addForwardedHeaders(HttpExchange exchange, HttpURLConnection conn) {
+ String remoteAddr = exchange.getRemoteAddress().getAddress().getHostAddress();
+ String xff = exchange.getRequestHeaders().getFirst("X-Forwarded-For");
+
+ if (StringUtils.isNotBlank(xff)) {
+ conn.setRequestProperty("X-Forwarded-For", xff + ", " + remoteAddr);
+ } else {
+ conn.setRequestProperty("X-Forwarded-For", remoteAddr);
+ }
+
+ conn.setRequestProperty("X-Forwarded-Proto", "http");
+ }
+
+ private void copyResponseBody(HttpExchange exchange, HttpURLConnection conn, int responseCode) throws IOException {
+ String transferEncoding = conn.getHeaderField("Transfer-Encoding");
+ String contentLengthStr = conn.getHeaderField("Content-Length");
+
+ if (transferEncoding != null && transferEncoding.equalsIgnoreCase("chunked")) {
+ exchange.sendResponseHeaders(responseCode, 0);
+ } else {
+ long contentLength = Optional.ofNullable(contentLengthStr).map(Long::parseLong).orElse(0L);
+ exchange.sendResponseHeaders(responseCode, contentLength);
+ }
+
+ try (InputStream is = responseCode >= 400 ? conn.getErrorStream() : conn.getInputStream();
+ OutputStream os = exchange.getResponseBody()) {
+ if (is != null) {
+ IOUtils.copyLarge(is, os, IO_BUFFER.get());
+ os.flush();
+ }
+ }
+ }
+
+ private HttpURLConnection createConnection(HttpExchange exchange, String target) throws IOException {
+ HttpURLConnection conn = (HttpURLConnection) new URL(target).openConnection();
+ String requestMethod = exchange.getRequestMethod();
+ conn.setRequestMethod(requestMethod);
+ conn.setConnectTimeout(CONNECT_TIMEOUT_MS);
+ conn.setReadTimeout(READ_TIMEOUT_MS);
+ return conn;
+ }
+
+ private void copyRequestBody(HttpExchange exchange, HttpURLConnection conn) throws IOException {
+ String method = exchange.getRequestMethod().toLowerCase();
+ if (method.equals("post") || method.equals("put") || method.equals("patch") || method.equals("delete")) {
+ String contentLength = exchange.getRequestHeaders().getFirst("Content-Length");
+ conn.setDoOutput(true);
+ if (StringUtils.isNotBlank(contentLength)) {
+ conn.setFixedLengthStreamingMode(Long.parseLong(contentLength));
+ } else {
+ conn.setChunkedStreamingMode(0);
+ }
+ try (InputStream reqBody = exchange.getRequestBody();
+ OutputStream os = conn.getOutputStream()) {
+ IOUtils.copy(reqBody, os);
+ }
+ }
+ }
+
+ private void copyRequestHeaders(HttpExchange exchange, HttpURLConnection conn) {
+ for (String headerName : exchange.getRequestHeaders().keySet()) {
+ if (isHopByHopHeader(headerName)) {
+ continue;
+ }
+
+ // Skip Expect header to avoid 100-continue issue
+ if ("Expect".equalsIgnoreCase(headerName)) {
+ continue;
+ }
+
+ List<String> headerValues = exchange.getRequestHeaders().get(headerName);
+ if (headerValues == null || headerValues.isEmpty()) {
+ continue;
+ }
+
+ for (String value : headerValues) {
+ // Convert header names to lowercase because HttpServer capitalizes the first letter,
+ // but s3g expects lowercase header names.
+ conn.addRequestProperty(headerName.toLowerCase(), value);
+ }
Review Comment:
Any issues if we let this be? I think for this if anything fails, we might need to handle it in S3G instead? Currently, we use `LowerCaseStringMap#fromHeaderMap` which is forcing the key to be lower case during signature processing.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-11679. Support multiple S3Gs in MiniOzoneCluster [ozone]
Posted by "ivandika3 (via GitHub)" <gi...@apache.org>.
ivandika3 commented on PR #8733:
URL: https://github.com/apache/ozone/pull/8733#issuecomment-3066611512
Thanks @hevinhsu for the improvement and @jojochuang @peterxcli for the reviews.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-11679. Support multiple S3Gs in MiniOzoneCluster [ozone]
Posted by "hevinhsu (via GitHub)" <gi...@apache.org>.
hevinhsu commented on code in PR #8733:
URL: https://github.com/apache/ozone/pull/8733#discussion_r2184142949
##########
hadoop-ozone/integration-test-s3/src/test/java/org/apache/hadoop/ozone/s3/ProxyServer.java:
##########
@@ -0,0 +1,251 @@
+/*
+ * 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.hadoop.ozone.s3;
+
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpHandler;
+import com.sun.net.httpserver.HttpServer;
Review Comment:
I’ve tried using the Hadoop HttpServer ([BaseHttpServer](https://github.com/apache/ozone/blob/master/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/server/http/BaseHttpServer.java)) for this implementation. However, it requires providing some additional configuration, such as web.xml.
To make it easier to set up an HTTP proxy server without extra configuration, I decided to use the standard library (com.sun.net.httpserver) instead.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-11679. Support multiple S3Gs in MiniOzoneCluster [ozone]
Posted by "ivandika3 (via GitHub)" <gi...@apache.org>.
ivandika3 merged PR #8733:
URL: https://github.com/apache/ozone/pull/8733
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-11679. Support multiple S3Gs in MiniOzoneCluster [ozone]
Posted by "ivandika3 (via GitHub)" <gi...@apache.org>.
ivandika3 commented on code in PR #8733:
URL: https://github.com/apache/ozone/pull/8733#discussion_r2184503663
##########
hadoop-ozone/integration-test-s3/src/test/java/org/apache/hadoop/ozone/s3/ProxyServer.java:
##########
@@ -0,0 +1,251 @@
+/*
+ * 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.hadoop.ozone.s3;
+
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpHandler;
+import com.sun.net.httpserver.HttpServer;
Review Comment:
+1 in using the `com.sun.net.httpserver` since it's minimalistic and do not require more dependencies. I checked `com.sun.net.httpserver` should not have issues on other JDKs so it should be safe (https://dev.to/thokuest/is-it-okay-to-use-comsunnethttpserverhttpserver-24il).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-11679. Support multiple S3Gs in MiniOzoneCluster [ozone]
Posted by "ivandika3 (via GitHub)" <gi...@apache.org>.
ivandika3 commented on code in PR #8733:
URL: https://github.com/apache/ozone/pull/8733#discussion_r2184465211
##########
hadoop-ozone/integration-test-s3/src/test/java/org/apache/hadoop/ozone/s3/ProxyServer.java:
##########
@@ -0,0 +1,251 @@
+/*
+ * 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.hadoop.ozone.s3;
+
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpHandler;
+import com.sun.net.httpserver.HttpServer;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.HttpURLConnection;
+import java.net.InetSocketAddress;
+import java.net.URL;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import org.apache.commons.io.IOUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * ProxyServer acts as a reverse proxy for S3G endpoints.
+ * It forwards requests to the selected S3G endpoint based on the load balancing strategy.
+ */
+public class ProxyServer {
+ private static final Logger LOG = LoggerFactory.getLogger(ProxyServer.class);
+ private final List<String> s3gEndpoints;
+ private final LoadBalanceStrategy loadBalanceStrategy;
+ private final HttpServer server;
+ private final String host;
+ private final int port;
+ private final ExecutorService executor;
+
+ // Add timeout configurations for better handling of large file operations
+ private static final int CONNECT_TIMEOUT_MS = 60000; // 60 seconds
+ private static final int READ_TIMEOUT_MS = 300000; // 5 minutes
+
+ private static final int BUFFER_SIZE = 64 * 1024; // 64KB
+ private static final ThreadLocal<byte[]> IO_BUFFER = ThreadLocal.withInitial(() -> new byte[BUFFER_SIZE]);
+
+ public ProxyServer(List<String> s3gEndpoints, String host, int proxyPort) throws Exception {
+ this(s3gEndpoints, host, proxyPort, new RoundRobinStrategy());
+ }
+
+ public ProxyServer(List<String> s3gEndpoints, String host, int proxyPort,
+ LoadBalanceStrategy loadBalanceStrategy) throws Exception {
+ this.s3gEndpoints = s3gEndpoints;
+ this.loadBalanceStrategy = loadBalanceStrategy;
+ this.host = host;
+ this.port = proxyPort;
+ server = HttpServer.create(new InetSocketAddress(host, proxyPort), 0);
+ server.createContext("/", new ProxyHandler());
+
+ this.executor = Executors.newCachedThreadPool();
+ server.setExecutor(executor);
+
+ LOG.info("ProxyServer initialized with endpoints: {}", s3gEndpoints);
+ LOG.info("Load balance strategy: {}", loadBalanceStrategy.getClass().getSimpleName());
+ LOG.info("Timeout settings - Connect: {}ms, Read: {}ms", CONNECT_TIMEOUT_MS, READ_TIMEOUT_MS);
+ }
+
+ public void start() {
+ server.start();
+ LOG.info("Proxy started on http://{}:{}", host, port);
+
+ }
+
+ public void stop() {
+ server.stop(0);
+ executor.shutdownNow();
+ LOG.info("Proxy stopped on http://{}:{}", host, port);
+ }
+
+ private class ProxyHandler implements HttpHandler {
+ @Override
+ public void handle(HttpExchange exchange) throws IOException {
+ HttpURLConnection conn = null;
+ try {
+ String target = loadBalanceStrategy.selectEndpoint(s3gEndpoints) + exchange.getRequestURI().toString();
+ conn = createConnection(exchange, target);
+ String requestMethod = conn.getRequestMethod();
+
+ LOG.info("Received request and forwarding request to [{}] {}", requestMethod, target);
+
+ copyRequestHeaders(exchange, conn);
+ addForwardedHeaders(exchange, conn);
+ copyRequestBody(exchange, conn);
+
+ int responseCode = conn.getResponseCode();
+ copyResponseHeaders(exchange, conn);
+
+ if (requestMethod.equals("HEAD") || responseCode == HttpURLConnection.HTTP_NO_CONTENT) {
+ exchange.sendResponseHeaders(responseCode, 0);
+ return;
+ }
+
+ copyResponseBody(exchange, conn, responseCode);
+
+ } catch (Exception e) {
+ LOG.error("Error forwarding request to S3G endpoint", e);
+ sendErrorResponse(exchange, e);
+ } finally {
+ closeResources(conn, exchange);
+ }
+ }
+
+ private void sendErrorResponse(HttpExchange exchange, Exception e) {
+ try {
+ exchange.sendResponseHeaders(502, 0);
+ try (OutputStream os = exchange.getResponseBody()) {
+ os.write(("Proxy error: " + e.getMessage()).getBytes(StandardCharsets.UTF_8));
+ }
+ } catch (IOException ioException) {
+ LOG.error("Failed to send error response", ioException);
+ }
+ }
+
+ private void closeResources(HttpURLConnection conn, HttpExchange exchange) {
+ if (conn != null) {
+ conn.disconnect();
+ }
+ exchange.close();
+ }
+
+ private void addForwardedHeaders(HttpExchange exchange, HttpURLConnection conn) {
+ String remoteAddr = exchange.getRemoteAddress().getAddress().getHostAddress();
+ String xff = exchange.getRequestHeaders().getFirst("X-Forwarded-For");
+
+ if (StringUtils.isNotBlank(xff)) {
+ conn.setRequestProperty("X-Forwarded-For", xff + ", " + remoteAddr);
+ } else {
+ conn.setRequestProperty("X-Forwarded-For", remoteAddr);
+ }
+
+ conn.setRequestProperty("X-Forwarded-Proto", "http");
+ }
+
+ private void copyResponseBody(HttpExchange exchange, HttpURLConnection conn, int responseCode) throws IOException {
+ String transferEncoding = conn.getHeaderField("Transfer-Encoding");
+ String contentLengthStr = conn.getHeaderField("Content-Length");
+
+ if (transferEncoding != null && transferEncoding.equalsIgnoreCase("chunked")) {
+ exchange.sendResponseHeaders(responseCode, 0);
+ } else {
+ long contentLength = Optional.ofNullable(contentLengthStr).map(Long::parseLong).orElse(0L);
+ exchange.sendResponseHeaders(responseCode, contentLength);
+ }
+
+ try (InputStream is = responseCode >= 400 ? conn.getErrorStream() : conn.getInputStream();
+ OutputStream os = exchange.getResponseBody()) {
+ if (is != null) {
+ IOUtils.copyLarge(is, os, IO_BUFFER.get());
+ os.flush();
+ }
+ }
+ }
+
+ private HttpURLConnection createConnection(HttpExchange exchange, String target) throws IOException {
+ HttpURLConnection conn = (HttpURLConnection) new URL(target).openConnection();
+ String requestMethod = exchange.getRequestMethod();
+ conn.setRequestMethod(requestMethod);
+ conn.setConnectTimeout(CONNECT_TIMEOUT_MS);
+ conn.setReadTimeout(READ_TIMEOUT_MS);
+ return conn;
+ }
+
+ private void copyRequestBody(HttpExchange exchange, HttpURLConnection conn) throws IOException {
+ String method = exchange.getRequestMethod().toLowerCase();
+ if (method.equals("post") || method.equals("put") || method.equals("patch") || method.equals("delete")) {
+ String contentLength = exchange.getRequestHeaders().getFirst("Content-Length");
+ conn.setDoOutput(true);
+ if (StringUtils.isNotBlank(contentLength)) {
+ conn.setFixedLengthStreamingMode(Long.parseLong(contentLength));
+ } else {
+ conn.setChunkedStreamingMode(0);
+ }
+ try (InputStream reqBody = exchange.getRequestBody();
+ OutputStream os = conn.getOutputStream()) {
+ IOUtils.copy(reqBody, os);
+ }
+ }
+ }
+
+ private void copyRequestHeaders(HttpExchange exchange, HttpURLConnection conn) {
+ for (String headerName : exchange.getRequestHeaders().keySet()) {
+ if (isHopByHopHeader(headerName)) {
+ continue;
+ }
+
+ // Skip Expect header to avoid 100-continue issue
+ if ("Expect".equalsIgnoreCase(headerName)) {
+ continue;
+ }
+
+ List<String> headerValues = exchange.getRequestHeaders().get(headerName);
+ if (headerValues == null || headerValues.isEmpty()) {
+ continue;
+ }
+
+ for (String value : headerValues) {
+ // Convert header names to lowercase because HttpServer capitalizes the first letter,
+ // but s3g expects lowercase header names.
+ conn.addRequestProperty(headerName.toLowerCase(), value);
+ }
+ }
+ }
+ }
+
+ private void copyResponseHeaders(HttpExchange exchange, HttpURLConnection conn) {
+ for (String headerName : conn.getHeaderFields().keySet()) {
+ if (headerName == null || isHopByHopHeader(headerName)) {
+ continue;
+ }
+
+ List<String> headerValues = conn.getHeaderFields().get(headerName);
+ if (headerValues == null) {
+ continue;
+ }
+
+ for (String headerValue : headerValues) {
+ exchange.getResponseHeaders().add(headerName, headerValue);
+ }
+ }
+ }
+
+ private boolean isHopByHopHeader(String header) {
+ String lowerName = header.toLowerCase();
+ return "connection".equals(lowerName) || "keep-alive".equals(lowerName)
+ || "proxy-authenticate".equals(lowerName) || "proxy-authorization".equals(lowerName)
+ || "te".equals(lowerName) || "trailers".equals(lowerName)
+ || "transfer-encoding".equals(lowerName) || "upgrade".equals(lowerName);
Review Comment:
We had a previous issues where `transfer-encoding` and `expect` was included in the signed headers (atlhough it is a hop-by-hop headers), but seems to have been dropped by the proxy. Triggering the following exceptions.
```
InternalError: Header expect not present in request but requested to be signed.\n\tstatus code: 500, request id: 28c2d896-d04c-4068-b854-9b036c03671f, host id: mQxBdH9Q8loGf
```
However, if it's passing on the current AWS SDK tests, I'm ok with dropping these request headers.
Nit: Can add some documentation link regarding hop-by-hop headers.
##########
hadoop-ozone/integration-test-s3/src/test/java/org/apache/hadoop/ozone/s3/ProxyServer.java:
##########
@@ -0,0 +1,251 @@
+/*
+ * 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.hadoop.ozone.s3;
+
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpHandler;
+import com.sun.net.httpserver.HttpServer;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.HttpURLConnection;
+import java.net.InetSocketAddress;
+import java.net.URL;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import org.apache.commons.io.IOUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * ProxyServer acts as a reverse proxy for S3G endpoints.
+ * It forwards requests to the selected S3G endpoint based on the load balancing strategy.
+ */
+public class ProxyServer {
+ private static final Logger LOG = LoggerFactory.getLogger(ProxyServer.class);
+ private final List<String> s3gEndpoints;
+ private final LoadBalanceStrategy loadBalanceStrategy;
+ private final HttpServer server;
+ private final String host;
+ private final int port;
+ private final ExecutorService executor;
+
+ // Add timeout configurations for better handling of large file operations
+ private static final int CONNECT_TIMEOUT_MS = 60000; // 60 seconds
+ private static final int READ_TIMEOUT_MS = 300000; // 5 minutes
+
+ private static final int BUFFER_SIZE = 64 * 1024; // 64KB
+ private static final ThreadLocal<byte[]> IO_BUFFER = ThreadLocal.withInitial(() -> new byte[BUFFER_SIZE]);
+
+ public ProxyServer(List<String> s3gEndpoints, String host, int proxyPort) throws Exception {
+ this(s3gEndpoints, host, proxyPort, new RoundRobinStrategy());
+ }
+
+ public ProxyServer(List<String> s3gEndpoints, String host, int proxyPort,
+ LoadBalanceStrategy loadBalanceStrategy) throws Exception {
+ this.s3gEndpoints = s3gEndpoints;
+ this.loadBalanceStrategy = loadBalanceStrategy;
+ this.host = host;
+ this.port = proxyPort;
+ server = HttpServer.create(new InetSocketAddress(host, proxyPort), 0);
+ server.createContext("/", new ProxyHandler());
+
+ this.executor = Executors.newCachedThreadPool();
+ server.setExecutor(executor);
+
+ LOG.info("ProxyServer initialized with endpoints: {}", s3gEndpoints);
+ LOG.info("Load balance strategy: {}", loadBalanceStrategy.getClass().getSimpleName());
+ LOG.info("Timeout settings - Connect: {}ms, Read: {}ms", CONNECT_TIMEOUT_MS, READ_TIMEOUT_MS);
+ }
+
+ public void start() {
+ server.start();
+ LOG.info("Proxy started on http://{}:{}", host, port);
+
+ }
+
+ public void stop() {
+ server.stop(0);
+ executor.shutdownNow();
+ LOG.info("Proxy stopped on http://{}:{}", host, port);
+ }
+
+ private class ProxyHandler implements HttpHandler {
+ @Override
+ public void handle(HttpExchange exchange) throws IOException {
+ HttpURLConnection conn = null;
+ try {
+ String target = loadBalanceStrategy.selectEndpoint(s3gEndpoints) + exchange.getRequestURI().toString();
+ conn = createConnection(exchange, target);
+ String requestMethod = conn.getRequestMethod();
+
+ LOG.info("Received request and forwarding request to [{}] {}", requestMethod, target);
+
+ copyRequestHeaders(exchange, conn);
+ addForwardedHeaders(exchange, conn);
+ copyRequestBody(exchange, conn);
+
+ int responseCode = conn.getResponseCode();
+ copyResponseHeaders(exchange, conn);
+
+ if (requestMethod.equals("HEAD") || responseCode == HttpURLConnection.HTTP_NO_CONTENT) {
+ exchange.sendResponseHeaders(responseCode, 0);
+ return;
+ }
+
+ copyResponseBody(exchange, conn, responseCode);
+
+ } catch (Exception e) {
+ LOG.error("Error forwarding request to S3G endpoint", e);
+ sendErrorResponse(exchange, e);
+ } finally {
+ closeResources(conn, exchange);
+ }
+ }
+
+ private void sendErrorResponse(HttpExchange exchange, Exception e) {
+ try {
+ exchange.sendResponseHeaders(502, 0);
+ try (OutputStream os = exchange.getResponseBody()) {
+ os.write(("Proxy error: " + e.getMessage()).getBytes(StandardCharsets.UTF_8));
+ }
+ } catch (IOException ioException) {
+ LOG.error("Failed to send error response", ioException);
+ }
+ }
+
+ private void closeResources(HttpURLConnection conn, HttpExchange exchange) {
+ if (conn != null) {
+ conn.disconnect();
+ }
+ exchange.close();
+ }
+
+ private void addForwardedHeaders(HttpExchange exchange, HttpURLConnection conn) {
+ String remoteAddr = exchange.getRemoteAddress().getAddress().getHostAddress();
+ String xff = exchange.getRequestHeaders().getFirst("X-Forwarded-For");
+
+ if (StringUtils.isNotBlank(xff)) {
+ conn.setRequestProperty("X-Forwarded-For", xff + ", " + remoteAddr);
+ } else {
+ conn.setRequestProperty("X-Forwarded-For", remoteAddr);
+ }
+
+ conn.setRequestProperty("X-Forwarded-Proto", "http");
+ }
+
+ private void copyResponseBody(HttpExchange exchange, HttpURLConnection conn, int responseCode) throws IOException {
+ String transferEncoding = conn.getHeaderField("Transfer-Encoding");
+ String contentLengthStr = conn.getHeaderField("Content-Length");
+
+ if (transferEncoding != null && transferEncoding.equalsIgnoreCase("chunked")) {
+ exchange.sendResponseHeaders(responseCode, 0);
+ } else {
+ long contentLength = Optional.ofNullable(contentLengthStr).map(Long::parseLong).orElse(0L);
+ exchange.sendResponseHeaders(responseCode, contentLength);
+ }
+
+ try (InputStream is = responseCode >= 400 ? conn.getErrorStream() : conn.getInputStream();
+ OutputStream os = exchange.getResponseBody()) {
+ if (is != null) {
+ IOUtils.copyLarge(is, os, IO_BUFFER.get());
+ os.flush();
+ }
+ }
+ }
+
+ private HttpURLConnection createConnection(HttpExchange exchange, String target) throws IOException {
+ HttpURLConnection conn = (HttpURLConnection) new URL(target).openConnection();
+ String requestMethod = exchange.getRequestMethod();
+ conn.setRequestMethod(requestMethod);
+ conn.setConnectTimeout(CONNECT_TIMEOUT_MS);
+ conn.setReadTimeout(READ_TIMEOUT_MS);
+ return conn;
+ }
+
+ private void copyRequestBody(HttpExchange exchange, HttpURLConnection conn) throws IOException {
+ String method = exchange.getRequestMethod().toLowerCase();
+ if (method.equals("post") || method.equals("put") || method.equals("patch") || method.equals("delete")) {
+ String contentLength = exchange.getRequestHeaders().getFirst("Content-Length");
+ conn.setDoOutput(true);
+ if (StringUtils.isNotBlank(contentLength)) {
+ conn.setFixedLengthStreamingMode(Long.parseLong(contentLength));
+ } else {
+ conn.setChunkedStreamingMode(0);
+ }
+ try (InputStream reqBody = exchange.getRequestBody();
+ OutputStream os = conn.getOutputStream()) {
+ IOUtils.copy(reqBody, os);
+ }
+ }
+ }
+
+ private void copyRequestHeaders(HttpExchange exchange, HttpURLConnection conn) {
+ for (String headerName : exchange.getRequestHeaders().keySet()) {
+ if (isHopByHopHeader(headerName)) {
+ continue;
+ }
+
+ // Skip Expect header to avoid 100-continue issue
+ if ("Expect".equalsIgnoreCase(headerName)) {
+ continue;
+ }
Review Comment:
Any issue if this is removed? From my research in https://issues.apache.org/jira/browse/HDDS-12842, it seems 100-Continue is handled automatically by S3G Jetty server.
##########
hadoop-ozone/integration-test-s3/src/test/java/org/apache/hadoop/ozone/s3/ProxyServer.java:
##########
@@ -0,0 +1,251 @@
+/*
+ * 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.hadoop.ozone.s3;
+
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpHandler;
+import com.sun.net.httpserver.HttpServer;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.HttpURLConnection;
+import java.net.InetSocketAddress;
+import java.net.URL;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import org.apache.commons.io.IOUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * ProxyServer acts as a reverse proxy for S3G endpoints.
+ * It forwards requests to the selected S3G endpoint based on the load balancing strategy.
+ */
+public class ProxyServer {
+ private static final Logger LOG = LoggerFactory.getLogger(ProxyServer.class);
+ private final List<String> s3gEndpoints;
+ private final LoadBalanceStrategy loadBalanceStrategy;
+ private final HttpServer server;
+ private final String host;
+ private final int port;
+ private final ExecutorService executor;
+
+ // Add timeout configurations for better handling of large file operations
+ private static final int CONNECT_TIMEOUT_MS = 60000; // 60 seconds
+ private static final int READ_TIMEOUT_MS = 300000; // 5 minutes
+
+ private static final int BUFFER_SIZE = 64 * 1024; // 64KB
+ private static final ThreadLocal<byte[]> IO_BUFFER = ThreadLocal.withInitial(() -> new byte[BUFFER_SIZE]);
+
+ public ProxyServer(List<String> s3gEndpoints, String host, int proxyPort) throws Exception {
+ this(s3gEndpoints, host, proxyPort, new RoundRobinStrategy());
+ }
+
+ public ProxyServer(List<String> s3gEndpoints, String host, int proxyPort,
+ LoadBalanceStrategy loadBalanceStrategy) throws Exception {
+ this.s3gEndpoints = s3gEndpoints;
+ this.loadBalanceStrategy = loadBalanceStrategy;
+ this.host = host;
+ this.port = proxyPort;
+ server = HttpServer.create(new InetSocketAddress(host, proxyPort), 0);
+ server.createContext("/", new ProxyHandler());
+
+ this.executor = Executors.newCachedThreadPool();
+ server.setExecutor(executor);
+
+ LOG.info("ProxyServer initialized with endpoints: {}", s3gEndpoints);
+ LOG.info("Load balance strategy: {}", loadBalanceStrategy.getClass().getSimpleName());
+ LOG.info("Timeout settings - Connect: {}ms, Read: {}ms", CONNECT_TIMEOUT_MS, READ_TIMEOUT_MS);
+ }
+
+ public void start() {
+ server.start();
+ LOG.info("Proxy started on http://{}:{}", host, port);
+
+ }
+
+ public void stop() {
+ server.stop(0);
+ executor.shutdownNow();
+ LOG.info("Proxy stopped on http://{}:{}", host, port);
+ }
+
+ private class ProxyHandler implements HttpHandler {
+ @Override
+ public void handle(HttpExchange exchange) throws IOException {
+ HttpURLConnection conn = null;
+ try {
+ String target = loadBalanceStrategy.selectEndpoint(s3gEndpoints) + exchange.getRequestURI().toString();
+ conn = createConnection(exchange, target);
+ String requestMethod = conn.getRequestMethod();
+
+ LOG.info("Received request and forwarding request to [{}] {}", requestMethod, target);
+
+ copyRequestHeaders(exchange, conn);
+ addForwardedHeaders(exchange, conn);
+ copyRequestBody(exchange, conn);
+
+ int responseCode = conn.getResponseCode();
+ copyResponseHeaders(exchange, conn);
+
+ if (requestMethod.equals("HEAD") || responseCode == HttpURLConnection.HTTP_NO_CONTENT) {
+ exchange.sendResponseHeaders(responseCode, 0);
+ return;
+ }
+
+ copyResponseBody(exchange, conn, responseCode);
+
+ } catch (Exception e) {
+ LOG.error("Error forwarding request to S3G endpoint", e);
+ sendErrorResponse(exchange, e);
+ } finally {
+ closeResources(conn, exchange);
+ }
+ }
+
+ private void sendErrorResponse(HttpExchange exchange, Exception e) {
+ try {
+ exchange.sendResponseHeaders(502, 0);
+ try (OutputStream os = exchange.getResponseBody()) {
+ os.write(("Proxy error: " + e.getMessage()).getBytes(StandardCharsets.UTF_8));
+ }
+ } catch (IOException ioException) {
+ LOG.error("Failed to send error response", ioException);
+ }
+ }
+
+ private void closeResources(HttpURLConnection conn, HttpExchange exchange) {
+ if (conn != null) {
+ conn.disconnect();
+ }
+ exchange.close();
+ }
+
+ private void addForwardedHeaders(HttpExchange exchange, HttpURLConnection conn) {
+ String remoteAddr = exchange.getRemoteAddress().getAddress().getHostAddress();
+ String xff = exchange.getRequestHeaders().getFirst("X-Forwarded-For");
+
+ if (StringUtils.isNotBlank(xff)) {
+ conn.setRequestProperty("X-Forwarded-For", xff + ", " + remoteAddr);
+ } else {
+ conn.setRequestProperty("X-Forwarded-For", remoteAddr);
+ }
+
+ conn.setRequestProperty("X-Forwarded-Proto", "http");
+ }
+
+ private void copyResponseBody(HttpExchange exchange, HttpURLConnection conn, int responseCode) throws IOException {
+ String transferEncoding = conn.getHeaderField("Transfer-Encoding");
+ String contentLengthStr = conn.getHeaderField("Content-Length");
+
+ if (transferEncoding != null && transferEncoding.equalsIgnoreCase("chunked")) {
+ exchange.sendResponseHeaders(responseCode, 0);
+ } else {
+ long contentLength = Optional.ofNullable(contentLengthStr).map(Long::parseLong).orElse(0L);
+ exchange.sendResponseHeaders(responseCode, contentLength);
+ }
+
+ try (InputStream is = responseCode >= 400 ? conn.getErrorStream() : conn.getInputStream();
+ OutputStream os = exchange.getResponseBody()) {
+ if (is != null) {
+ IOUtils.copyLarge(is, os, IO_BUFFER.get());
+ os.flush();
+ }
+ }
+ }
+
+ private HttpURLConnection createConnection(HttpExchange exchange, String target) throws IOException {
+ HttpURLConnection conn = (HttpURLConnection) new URL(target).openConnection();
+ String requestMethod = exchange.getRequestMethod();
+ conn.setRequestMethod(requestMethod);
+ conn.setConnectTimeout(CONNECT_TIMEOUT_MS);
+ conn.setReadTimeout(READ_TIMEOUT_MS);
+ return conn;
+ }
+
+ private void copyRequestBody(HttpExchange exchange, HttpURLConnection conn) throws IOException {
+ String method = exchange.getRequestMethod().toLowerCase();
+ if (method.equals("post") || method.equals("put") || method.equals("patch") || method.equals("delete")) {
+ String contentLength = exchange.getRequestHeaders().getFirst("Content-Length");
+ conn.setDoOutput(true);
+ if (StringUtils.isNotBlank(contentLength)) {
+ conn.setFixedLengthStreamingMode(Long.parseLong(contentLength));
+ } else {
+ conn.setChunkedStreamingMode(0);
+ }
+ try (InputStream reqBody = exchange.getRequestBody();
+ OutputStream os = conn.getOutputStream()) {
+ IOUtils.copy(reqBody, os);
+ }
+ }
+ }
+
+ private void copyRequestHeaders(HttpExchange exchange, HttpURLConnection conn) {
+ for (String headerName : exchange.getRequestHeaders().keySet()) {
+ if (isHopByHopHeader(headerName)) {
+ continue;
+ }
+
+ // Skip Expect header to avoid 100-continue issue
+ if ("Expect".equalsIgnoreCase(headerName)) {
+ continue;
+ }
+
+ List<String> headerValues = exchange.getRequestHeaders().get(headerName);
+ if (headerValues == null || headerValues.isEmpty()) {
+ continue;
+ }
+
+ for (String value : headerValues) {
+ // Convert header names to lowercase because HttpServer capitalizes the first letter,
+ // but s3g expects lowercase header names.
+ conn.addRequestProperty(headerName.toLowerCase(), value);
+ }
Review Comment:
I think for this if it's valid header keys, we might need to handle it in S3G instead? Currently, we use `LowerCaseStringMap#fromHeaderMap` which is forcing the key to be lower case during signature processing.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-11679. Support multiple S3Gs in MiniOzoneCluster [ozone]
Posted by "jojochuang (via GitHub)" <gi...@apache.org>.
jojochuang commented on code in PR #8733:
URL: https://github.com/apache/ozone/pull/8733#discussion_r2183255071
##########
hadoop-ozone/integration-test-s3/src/test/java/org/apache/hadoop/ozone/s3/ProxyServer.java:
##########
@@ -0,0 +1,251 @@
+/*
+ * 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.hadoop.ozone.s3;
+
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpHandler;
+import com.sun.net.httpserver.HttpServer;
Review Comment:
i have no experience with this implementation. But looks like it does the job. I usually go with a netty or jetty based implementation such as the Hadoop HttpServer.
##########
hadoop-ozone/integration-test-s3/src/test/java/org/apache/hadoop/ozone/s3/RoundRobinStrategy.java:
##########
@@ -0,0 +1,41 @@
+/*
+ * 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.hadoop.ozone.s3;
+
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * Round Robin load balancing strategy.
+ * Distributes requests evenly across all available endpoints in a circular manner.
+ */
+public class RoundRobinStrategy implements LoadBalanceStrategy {
Review Comment:
Round robin is stateless. For production deployments, S3G has session state (e.g., authentication, in-memory data), you may run into issues. For integration tests, likely not a problem.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-11679. Support multiple S3Gs in MiniOzoneCluster [ozone]
Posted by "ivandika3 (via GitHub)" <gi...@apache.org>.
ivandika3 commented on code in PR #8733:
URL: https://github.com/apache/ozone/pull/8733#discussion_r2184503663
##########
hadoop-ozone/integration-test-s3/src/test/java/org/apache/hadoop/ozone/s3/ProxyServer.java:
##########
@@ -0,0 +1,251 @@
+/*
+ * 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.hadoop.ozone.s3;
+
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpHandler;
+import com.sun.net.httpserver.HttpServer;
Review Comment:
+1 in using the `com.sun.net.httpserver` and not adding more dependencies. I checked `com.sun.net.httpserver` should not have issues on other JDKs so it should be safe (https://dev.to/thokuest/is-it-okay-to-use-comsunnethttpserverhttpserver-24il).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-11679. Support multiple S3Gs in MiniOzoneCluster [ozone]
Posted by "hevinhsu (via GitHub)" <gi...@apache.org>.
hevinhsu commented on PR #8733:
URL: https://github.com/apache/ozone/pull/8733#issuecomment-3061207574
> I found that Jetty has a ProxyServlet, it seems to require minimal configuration.
@peterxcli Thanks for the reference! It's really helpful for implementing a proxy server that only needs to handle URL rewriting and the 100-continue issue. This will save a lot of effort compared to building the proxy from scratch.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-11679. Support multiple S3Gs in MiniOzoneCluster [ozone]
Posted by "ivandika3 (via GitHub)" <gi...@apache.org>.
ivandika3 commented on code in PR #8733:
URL: https://github.com/apache/ozone/pull/8733#discussion_r2187146640
##########
hadoop-ozone/integration-test-s3/src/test/java/org/apache/hadoop/ozone/s3/ProxyServer.java:
##########
@@ -0,0 +1,251 @@
+/*
+ * 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.hadoop.ozone.s3;
+
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpHandler;
+import com.sun.net.httpserver.HttpServer;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.HttpURLConnection;
+import java.net.InetSocketAddress;
+import java.net.URL;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import org.apache.commons.io.IOUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * ProxyServer acts as a reverse proxy for S3G endpoints.
+ * It forwards requests to the selected S3G endpoint based on the load balancing strategy.
+ */
+public class ProxyServer {
+ private static final Logger LOG = LoggerFactory.getLogger(ProxyServer.class);
+ private final List<String> s3gEndpoints;
+ private final LoadBalanceStrategy loadBalanceStrategy;
+ private final HttpServer server;
+ private final String host;
+ private final int port;
+ private final ExecutorService executor;
+
+ // Add timeout configurations for better handling of large file operations
+ private static final int CONNECT_TIMEOUT_MS = 60000; // 60 seconds
+ private static final int READ_TIMEOUT_MS = 300000; // 5 minutes
+
+ private static final int BUFFER_SIZE = 64 * 1024; // 64KB
+ private static final ThreadLocal<byte[]> IO_BUFFER = ThreadLocal.withInitial(() -> new byte[BUFFER_SIZE]);
+
+ public ProxyServer(List<String> s3gEndpoints, String host, int proxyPort) throws Exception {
+ this(s3gEndpoints, host, proxyPort, new RoundRobinStrategy());
+ }
+
+ public ProxyServer(List<String> s3gEndpoints, String host, int proxyPort,
+ LoadBalanceStrategy loadBalanceStrategy) throws Exception {
+ this.s3gEndpoints = s3gEndpoints;
+ this.loadBalanceStrategy = loadBalanceStrategy;
+ this.host = host;
+ this.port = proxyPort;
+ server = HttpServer.create(new InetSocketAddress(host, proxyPort), 0);
+ server.createContext("/", new ProxyHandler());
+
+ this.executor = Executors.newCachedThreadPool();
+ server.setExecutor(executor);
+
+ LOG.info("ProxyServer initialized with endpoints: {}", s3gEndpoints);
+ LOG.info("Load balance strategy: {}", loadBalanceStrategy.getClass().getSimpleName());
+ LOG.info("Timeout settings - Connect: {}ms, Read: {}ms", CONNECT_TIMEOUT_MS, READ_TIMEOUT_MS);
+ }
+
+ public void start() {
+ server.start();
+ LOG.info("Proxy started on http://{}:{}", host, port);
+
+ }
+
+ public void stop() {
+ server.stop(0);
+ executor.shutdownNow();
+ LOG.info("Proxy stopped on http://{}:{}", host, port);
+ }
+
+ private class ProxyHandler implements HttpHandler {
+ @Override
+ public void handle(HttpExchange exchange) throws IOException {
+ HttpURLConnection conn = null;
+ try {
+ String target = loadBalanceStrategy.selectEndpoint(s3gEndpoints) + exchange.getRequestURI().toString();
+ conn = createConnection(exchange, target);
+ String requestMethod = conn.getRequestMethod();
+
+ LOG.info("Received request and forwarding request to [{}] {}", requestMethod, target);
+
+ copyRequestHeaders(exchange, conn);
+ addForwardedHeaders(exchange, conn);
+ copyRequestBody(exchange, conn);
+
+ int responseCode = conn.getResponseCode();
+ copyResponseHeaders(exchange, conn);
+
+ if (requestMethod.equals("HEAD") || responseCode == HttpURLConnection.HTTP_NO_CONTENT) {
+ exchange.sendResponseHeaders(responseCode, 0);
+ return;
+ }
+
+ copyResponseBody(exchange, conn, responseCode);
+
+ } catch (Exception e) {
+ LOG.error("Error forwarding request to S3G endpoint", e);
+ sendErrorResponse(exchange, e);
+ } finally {
+ closeResources(conn, exchange);
+ }
+ }
+
+ private void sendErrorResponse(HttpExchange exchange, Exception e) {
+ try {
+ exchange.sendResponseHeaders(502, 0);
+ try (OutputStream os = exchange.getResponseBody()) {
+ os.write(("Proxy error: " + e.getMessage()).getBytes(StandardCharsets.UTF_8));
+ }
+ } catch (IOException ioException) {
+ LOG.error("Failed to send error response", ioException);
+ }
+ }
+
+ private void closeResources(HttpURLConnection conn, HttpExchange exchange) {
+ if (conn != null) {
+ conn.disconnect();
+ }
+ exchange.close();
+ }
+
+ private void addForwardedHeaders(HttpExchange exchange, HttpURLConnection conn) {
+ String remoteAddr = exchange.getRemoteAddress().getAddress().getHostAddress();
+ String xff = exchange.getRequestHeaders().getFirst("X-Forwarded-For");
+
+ if (StringUtils.isNotBlank(xff)) {
+ conn.setRequestProperty("X-Forwarded-For", xff + ", " + remoteAddr);
+ } else {
+ conn.setRequestProperty("X-Forwarded-For", remoteAddr);
+ }
+
+ conn.setRequestProperty("X-Forwarded-Proto", "http");
+ }
+
+ private void copyResponseBody(HttpExchange exchange, HttpURLConnection conn, int responseCode) throws IOException {
+ String transferEncoding = conn.getHeaderField("Transfer-Encoding");
+ String contentLengthStr = conn.getHeaderField("Content-Length");
+
+ if (transferEncoding != null && transferEncoding.equalsIgnoreCase("chunked")) {
+ exchange.sendResponseHeaders(responseCode, 0);
+ } else {
+ long contentLength = Optional.ofNullable(contentLengthStr).map(Long::parseLong).orElse(0L);
+ exchange.sendResponseHeaders(responseCode, contentLength);
+ }
+
+ try (InputStream is = responseCode >= 400 ? conn.getErrorStream() : conn.getInputStream();
+ OutputStream os = exchange.getResponseBody()) {
+ if (is != null) {
+ IOUtils.copyLarge(is, os, IO_BUFFER.get());
+ os.flush();
+ }
+ }
+ }
+
+ private HttpURLConnection createConnection(HttpExchange exchange, String target) throws IOException {
+ HttpURLConnection conn = (HttpURLConnection) new URL(target).openConnection();
+ String requestMethod = exchange.getRequestMethod();
+ conn.setRequestMethod(requestMethod);
+ conn.setConnectTimeout(CONNECT_TIMEOUT_MS);
+ conn.setReadTimeout(READ_TIMEOUT_MS);
+ return conn;
+ }
+
+ private void copyRequestBody(HttpExchange exchange, HttpURLConnection conn) throws IOException {
+ String method = exchange.getRequestMethod().toLowerCase();
+ if (method.equals("post") || method.equals("put") || method.equals("patch") || method.equals("delete")) {
+ String contentLength = exchange.getRequestHeaders().getFirst("Content-Length");
+ conn.setDoOutput(true);
+ if (StringUtils.isNotBlank(contentLength)) {
+ conn.setFixedLengthStreamingMode(Long.parseLong(contentLength));
+ } else {
+ conn.setChunkedStreamingMode(0);
+ }
+ try (InputStream reqBody = exchange.getRequestBody();
+ OutputStream os = conn.getOutputStream()) {
+ IOUtils.copy(reqBody, os);
+ }
+ }
+ }
+
+ private void copyRequestHeaders(HttpExchange exchange, HttpURLConnection conn) {
+ for (String headerName : exchange.getRequestHeaders().keySet()) {
+ if (isHopByHopHeader(headerName)) {
+ continue;
+ }
+
+ // Skip Expect header to avoid 100-continue issue
+ if ("Expect".equalsIgnoreCase(headerName)) {
+ continue;
+ }
Review Comment:
@hevinhsu I see, good to know. My understanding is the 100-Continue contains empty response body and should be acknowledged by 100 Continue response. If S3G handles it normally, I imagine it might cause the S3G to upload an empty object.
However, I'm fine if this is working with the existing `PutObject` request.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-11679. Support multiple S3Gs in MiniOzoneCluster [ozone]
Posted by "peterxcli (via GitHub)" <gi...@apache.org>.
peterxcli commented on code in PR #8733:
URL: https://github.com/apache/ozone/pull/8733#discussion_r2187970289
##########
hadoop-ozone/integration-test-s3/src/test/java/org/apache/hadoop/ozone/s3/ProxyServer.java:
##########
@@ -0,0 +1,257 @@
+/*
+ * 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.hadoop.ozone.s3;
+
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpHandler;
+import com.sun.net.httpserver.HttpServer;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.HttpURLConnection;
+import java.net.InetSocketAddress;
+import java.net.URL;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import org.apache.commons.io.IOUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * ProxyServer acts as a reverse proxy for S3G endpoints.
+ * It forwards requests to the selected S3G endpoint based on the load balancing strategy.
+ */
+public class ProxyServer {
+ private static final Logger LOG = LoggerFactory.getLogger(ProxyServer.class);
+ private final List<String> s3gEndpoints;
+ private final LoadBalanceStrategy loadBalanceStrategy;
+ private final HttpServer server;
+ private final String host;
+ private final int port;
+ private final ExecutorService executor;
+
+ // Add timeout configurations for better handling of large file operations
+ private static final int CONNECT_TIMEOUT_MS = 60000; // 60 seconds
+ private static final int READ_TIMEOUT_MS = 300000; // 5 minutes
+
+ private static final int BUFFER_SIZE = 64 * 1024; // 64KB
+ private static final ThreadLocal<byte[]> IO_BUFFER = ThreadLocal.withInitial(() -> new byte[BUFFER_SIZE]);
+
+ public ProxyServer(List<String> s3gEndpoints, String host, int proxyPort) throws Exception {
+ this(s3gEndpoints, host, proxyPort, new RoundRobinStrategy());
+ }
+
+ public ProxyServer(List<String> s3gEndpoints, String host, int proxyPort,
+ LoadBalanceStrategy loadBalanceStrategy) throws Exception {
+ this.s3gEndpoints = s3gEndpoints;
+ this.loadBalanceStrategy = loadBalanceStrategy;
+ this.host = host;
+ this.port = proxyPort;
+ server = HttpServer.create(new InetSocketAddress(host, proxyPort), 0);
+ server.createContext("/", new ProxyHandler());
+
+ this.executor = Executors.newCachedThreadPool();
+ server.setExecutor(executor);
+
+ LOG.info("ProxyServer initialized with endpoints: {}", s3gEndpoints);
+ LOG.info("Load balance strategy: {}", loadBalanceStrategy.getClass().getSimpleName());
+ LOG.info("Timeout settings - Connect: {}ms, Read: {}ms", CONNECT_TIMEOUT_MS, READ_TIMEOUT_MS);
+ }
+
+ public void start() {
+ server.start();
+ LOG.info("Proxy started on http://{}:{}", host, port);
+
+ }
+
+ public void stop() {
+ server.stop(0);
+ executor.shutdownNow();
+ LOG.info("Proxy stopped on http://{}:{}", host, port);
+ }
+
+ private class ProxyHandler implements HttpHandler {
Review Comment:
Do you have any reference for this logic?
I'm not sure if we might be missing something here, even if the tests pass — it could still lead to errors in the future.
##########
hadoop-ozone/integration-test-s3/src/test/java/org/apache/hadoop/ozone/s3/ProxyServer.java:
##########
@@ -0,0 +1,251 @@
+/*
+ * 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.hadoop.ozone.s3;
+
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpHandler;
+import com.sun.net.httpserver.HttpServer;
Review Comment:
I found that Jetty has a `ProxyServlet`, it seems to require minimal configuration.
ref: https://gist.github.com/jponge/1752767
Please take a look.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-11679. Support multiple S3Gs in MiniOzoneCluster [ozone]
Posted by "hevinhsu (via GitHub)" <gi...@apache.org>.
hevinhsu commented on PR #8733:
URL: https://github.com/apache/ozone/pull/8733#issuecomment-3038597131
@ivandika3 Thank you for the feedback and suggestions.
Regarding the unit test, my initial thought is to add a unit test for RoundRobinStrategy, but I’m not sure if that would be sufficient.
As for merging with the S3G integration tests, do you mean copying the existing integration test classes and replacing S3GatewayService with MultiS3GatewayService?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-11679. Support multiple S3Gs in MiniOzoneCluster [ozone]
Posted by "hevinhsu (via GitHub)" <gi...@apache.org>.
hevinhsu commented on PR #8733:
URL: https://github.com/apache/ozone/pull/8733#issuecomment-3061217876
>Yes, we can add a simple unit test on the RoundRobinStrategy, although it should be straightforward. The more important test is to validate that the proxy actual routes to different S3Gs, maybe through using Spy or parsing the S3G audit logs.
@ivandika3 I couldn't find a reliable way to identify which S3G endpoint was actually called. I tried parsing the audit logs, but there was no identifier that could distinguish the specific endpoint. I also explored some HTTP endpoints that might expose this information, but couldn't find anything useful — though it's possible I might have missed something in all of these attempts.
If I overlooked anything, please feel free to let me know.
So instead, I started a simple HTTP server to verify the routing logic. Please take a look and let me know if this approach seems acceptable for validating the strategy.
>I'm OK with replacing the S3GatewayService to MultiS3GatewayService as long as the tests are fine.
Since the ProxyServer implementation now uses Jetty's ProxyServlet, which handles HTTP headers properly, I've replaced S3GatewayService with MultiS3GatewayService in the integration tests.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-11679. Support multiple S3Gs in MiniOzoneCluster [ozone]
Posted by "hevinhsu (via GitHub)" <gi...@apache.org>.
hevinhsu commented on PR #8733:
URL: https://github.com/apache/ozone/pull/8733#issuecomment-3066441592
@ivandika3 Nits fixed. PTAL, thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-11679. Support multiple S3Gs in MiniOzoneCluster [ozone]
Posted by "jojochuang (via GitHub)" <gi...@apache.org>.
jojochuang commented on PR #8733:
URL: https://github.com/apache/ozone/pull/8733#issuecomment-3066590709
Let's get this in. If there are follow-up comments we can address them later.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-11679. Support multiple S3Gs in MiniOzoneCluster [ozone]
Posted by "hevinhsu (via GitHub)" <gi...@apache.org>.
hevinhsu commented on PR #8733:
URL: https://github.com/apache/ozone/pull/8733#issuecomment-3066729489
Thanks @jojochuang @ivandika3 @peterxcli for the reviews!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-11679. Support multiple S3Gs in MiniOzoneCluster [ozone]
Posted by "ivandika3 (via GitHub)" <gi...@apache.org>.
ivandika3 commented on code in PR #8733:
URL: https://github.com/apache/ozone/pull/8733#discussion_r2184473765
##########
hadoop-ozone/integration-test-s3/src/test/java/org/apache/hadoop/ozone/s3/ProxyServer.java:
##########
@@ -0,0 +1,251 @@
+/*
+ * 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.hadoop.ozone.s3;
+
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpHandler;
+import com.sun.net.httpserver.HttpServer;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.HttpURLConnection;
+import java.net.InetSocketAddress;
+import java.net.URL;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import org.apache.commons.io.IOUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * ProxyServer acts as a reverse proxy for S3G endpoints.
+ * It forwards requests to the selected S3G endpoint based on the load balancing strategy.
+ */
+public class ProxyServer {
+ private static final Logger LOG = LoggerFactory.getLogger(ProxyServer.class);
+ private final List<String> s3gEndpoints;
+ private final LoadBalanceStrategy loadBalanceStrategy;
+ private final HttpServer server;
+ private final String host;
+ private final int port;
+ private final ExecutorService executor;
+
+ // Add timeout configurations for better handling of large file operations
+ private static final int CONNECT_TIMEOUT_MS = 60000; // 60 seconds
+ private static final int READ_TIMEOUT_MS = 300000; // 5 minutes
+
+ private static final int BUFFER_SIZE = 64 * 1024; // 64KB
+ private static final ThreadLocal<byte[]> IO_BUFFER = ThreadLocal.withInitial(() -> new byte[BUFFER_SIZE]);
+
+ public ProxyServer(List<String> s3gEndpoints, String host, int proxyPort) throws Exception {
+ this(s3gEndpoints, host, proxyPort, new RoundRobinStrategy());
+ }
+
+ public ProxyServer(List<String> s3gEndpoints, String host, int proxyPort,
+ LoadBalanceStrategy loadBalanceStrategy) throws Exception {
+ this.s3gEndpoints = s3gEndpoints;
+ this.loadBalanceStrategy = loadBalanceStrategy;
+ this.host = host;
+ this.port = proxyPort;
+ server = HttpServer.create(new InetSocketAddress(host, proxyPort), 0);
+ server.createContext("/", new ProxyHandler());
+
+ this.executor = Executors.newCachedThreadPool();
+ server.setExecutor(executor);
+
+ LOG.info("ProxyServer initialized with endpoints: {}", s3gEndpoints);
+ LOG.info("Load balance strategy: {}", loadBalanceStrategy.getClass().getSimpleName());
+ LOG.info("Timeout settings - Connect: {}ms, Read: {}ms", CONNECT_TIMEOUT_MS, READ_TIMEOUT_MS);
+ }
+
+ public void start() {
+ server.start();
+ LOG.info("Proxy started on http://{}:{}", host, port);
+
+ }
+
+ public void stop() {
+ server.stop(0);
+ executor.shutdownNow();
+ LOG.info("Proxy stopped on http://{}:{}", host, port);
+ }
+
+ private class ProxyHandler implements HttpHandler {
+ @Override
+ public void handle(HttpExchange exchange) throws IOException {
+ HttpURLConnection conn = null;
+ try {
+ String target = loadBalanceStrategy.selectEndpoint(s3gEndpoints) + exchange.getRequestURI().toString();
+ conn = createConnection(exchange, target);
+ String requestMethod = conn.getRequestMethod();
+
+ LOG.info("Received request and forwarding request to [{}] {}", requestMethod, target);
+
+ copyRequestHeaders(exchange, conn);
+ addForwardedHeaders(exchange, conn);
+ copyRequestBody(exchange, conn);
+
+ int responseCode = conn.getResponseCode();
+ copyResponseHeaders(exchange, conn);
+
+ if (requestMethod.equals("HEAD") || responseCode == HttpURLConnection.HTTP_NO_CONTENT) {
+ exchange.sendResponseHeaders(responseCode, 0);
+ return;
+ }
+
+ copyResponseBody(exchange, conn, responseCode);
+
+ } catch (Exception e) {
+ LOG.error("Error forwarding request to S3G endpoint", e);
+ sendErrorResponse(exchange, e);
+ } finally {
+ closeResources(conn, exchange);
+ }
+ }
+
+ private void sendErrorResponse(HttpExchange exchange, Exception e) {
+ try {
+ exchange.sendResponseHeaders(502, 0);
+ try (OutputStream os = exchange.getResponseBody()) {
+ os.write(("Proxy error: " + e.getMessage()).getBytes(StandardCharsets.UTF_8));
+ }
+ } catch (IOException ioException) {
+ LOG.error("Failed to send error response", ioException);
+ }
+ }
+
+ private void closeResources(HttpURLConnection conn, HttpExchange exchange) {
+ if (conn != null) {
+ conn.disconnect();
+ }
+ exchange.close();
+ }
+
+ private void addForwardedHeaders(HttpExchange exchange, HttpURLConnection conn) {
+ String remoteAddr = exchange.getRemoteAddress().getAddress().getHostAddress();
+ String xff = exchange.getRequestHeaders().getFirst("X-Forwarded-For");
+
+ if (StringUtils.isNotBlank(xff)) {
+ conn.setRequestProperty("X-Forwarded-For", xff + ", " + remoteAddr);
+ } else {
+ conn.setRequestProperty("X-Forwarded-For", remoteAddr);
+ }
+
+ conn.setRequestProperty("X-Forwarded-Proto", "http");
+ }
+
+ private void copyResponseBody(HttpExchange exchange, HttpURLConnection conn, int responseCode) throws IOException {
+ String transferEncoding = conn.getHeaderField("Transfer-Encoding");
+ String contentLengthStr = conn.getHeaderField("Content-Length");
+
+ if (transferEncoding != null && transferEncoding.equalsIgnoreCase("chunked")) {
+ exchange.sendResponseHeaders(responseCode, 0);
+ } else {
+ long contentLength = Optional.ofNullable(contentLengthStr).map(Long::parseLong).orElse(0L);
+ exchange.sendResponseHeaders(responseCode, contentLength);
+ }
+
+ try (InputStream is = responseCode >= 400 ? conn.getErrorStream() : conn.getInputStream();
+ OutputStream os = exchange.getResponseBody()) {
+ if (is != null) {
+ IOUtils.copyLarge(is, os, IO_BUFFER.get());
+ os.flush();
+ }
+ }
+ }
+
+ private HttpURLConnection createConnection(HttpExchange exchange, String target) throws IOException {
+ HttpURLConnection conn = (HttpURLConnection) new URL(target).openConnection();
+ String requestMethod = exchange.getRequestMethod();
+ conn.setRequestMethod(requestMethod);
+ conn.setConnectTimeout(CONNECT_TIMEOUT_MS);
+ conn.setReadTimeout(READ_TIMEOUT_MS);
+ return conn;
+ }
+
+ private void copyRequestBody(HttpExchange exchange, HttpURLConnection conn) throws IOException {
+ String method = exchange.getRequestMethod().toLowerCase();
+ if (method.equals("post") || method.equals("put") || method.equals("patch") || method.equals("delete")) {
+ String contentLength = exchange.getRequestHeaders().getFirst("Content-Length");
+ conn.setDoOutput(true);
+ if (StringUtils.isNotBlank(contentLength)) {
+ conn.setFixedLengthStreamingMode(Long.parseLong(contentLength));
+ } else {
+ conn.setChunkedStreamingMode(0);
+ }
+ try (InputStream reqBody = exchange.getRequestBody();
+ OutputStream os = conn.getOutputStream()) {
+ IOUtils.copy(reqBody, os);
+ }
+ }
+ }
+
+ private void copyRequestHeaders(HttpExchange exchange, HttpURLConnection conn) {
+ for (String headerName : exchange.getRequestHeaders().keySet()) {
+ if (isHopByHopHeader(headerName)) {
+ continue;
+ }
+
+ // Skip Expect header to avoid 100-continue issue
+ if ("Expect".equalsIgnoreCase(headerName)) {
+ continue;
+ }
+
+ List<String> headerValues = exchange.getRequestHeaders().get(headerName);
+ if (headerValues == null || headerValues.isEmpty()) {
+ continue;
+ }
+
+ for (String value : headerValues) {
+ // Convert header names to lowercase because HttpServer capitalizes the first letter,
+ // but s3g expects lowercase header names.
+ conn.addRequestProperty(headerName.toLowerCase(), value);
+ }
Review Comment:
Any issues if we don't lower case the header keys? I think for this if anything fails, we might need to handle it in S3G instead? Currently, we use `LowerCaseStringMap#fromHeaderMap` which is forcing the key to be lower case during signature processing.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-11679. Support multiple S3Gs in MiniOzoneCluster [ozone]
Posted by "hevinhsu (via GitHub)" <gi...@apache.org>.
hevinhsu commented on code in PR #8733:
URL: https://github.com/apache/ozone/pull/8733#discussion_r2186981954
##########
hadoop-ozone/integration-test-s3/src/test/java/org/apache/hadoop/ozone/s3/ProxyServer.java:
##########
@@ -0,0 +1,251 @@
+/*
+ * 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.hadoop.ozone.s3;
+
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpHandler;
+import com.sun.net.httpserver.HttpServer;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.HttpURLConnection;
+import java.net.InetSocketAddress;
+import java.net.URL;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import org.apache.commons.io.IOUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * ProxyServer acts as a reverse proxy for S3G endpoints.
+ * It forwards requests to the selected S3G endpoint based on the load balancing strategy.
+ */
+public class ProxyServer {
+ private static final Logger LOG = LoggerFactory.getLogger(ProxyServer.class);
+ private final List<String> s3gEndpoints;
+ private final LoadBalanceStrategy loadBalanceStrategy;
+ private final HttpServer server;
+ private final String host;
+ private final int port;
+ private final ExecutorService executor;
+
+ // Add timeout configurations for better handling of large file operations
+ private static final int CONNECT_TIMEOUT_MS = 60000; // 60 seconds
+ private static final int READ_TIMEOUT_MS = 300000; // 5 minutes
+
+ private static final int BUFFER_SIZE = 64 * 1024; // 64KB
+ private static final ThreadLocal<byte[]> IO_BUFFER = ThreadLocal.withInitial(() -> new byte[BUFFER_SIZE]);
+
+ public ProxyServer(List<String> s3gEndpoints, String host, int proxyPort) throws Exception {
+ this(s3gEndpoints, host, proxyPort, new RoundRobinStrategy());
+ }
+
+ public ProxyServer(List<String> s3gEndpoints, String host, int proxyPort,
+ LoadBalanceStrategy loadBalanceStrategy) throws Exception {
+ this.s3gEndpoints = s3gEndpoints;
+ this.loadBalanceStrategy = loadBalanceStrategy;
+ this.host = host;
+ this.port = proxyPort;
+ server = HttpServer.create(new InetSocketAddress(host, proxyPort), 0);
+ server.createContext("/", new ProxyHandler());
+
+ this.executor = Executors.newCachedThreadPool();
+ server.setExecutor(executor);
+
+ LOG.info("ProxyServer initialized with endpoints: {}", s3gEndpoints);
+ LOG.info("Load balance strategy: {}", loadBalanceStrategy.getClass().getSimpleName());
+ LOG.info("Timeout settings - Connect: {}ms, Read: {}ms", CONNECT_TIMEOUT_MS, READ_TIMEOUT_MS);
+ }
+
+ public void start() {
+ server.start();
+ LOG.info("Proxy started on http://{}:{}", host, port);
+
+ }
+
+ public void stop() {
+ server.stop(0);
+ executor.shutdownNow();
+ LOG.info("Proxy stopped on http://{}:{}", host, port);
+ }
+
+ private class ProxyHandler implements HttpHandler {
+ @Override
+ public void handle(HttpExchange exchange) throws IOException {
+ HttpURLConnection conn = null;
+ try {
+ String target = loadBalanceStrategy.selectEndpoint(s3gEndpoints) + exchange.getRequestURI().toString();
+ conn = createConnection(exchange, target);
+ String requestMethod = conn.getRequestMethod();
+
+ LOG.info("Received request and forwarding request to [{}] {}", requestMethod, target);
+
+ copyRequestHeaders(exchange, conn);
+ addForwardedHeaders(exchange, conn);
+ copyRequestBody(exchange, conn);
+
+ int responseCode = conn.getResponseCode();
+ copyResponseHeaders(exchange, conn);
+
+ if (requestMethod.equals("HEAD") || responseCode == HttpURLConnection.HTTP_NO_CONTENT) {
+ exchange.sendResponseHeaders(responseCode, 0);
+ return;
+ }
+
+ copyResponseBody(exchange, conn, responseCode);
+
+ } catch (Exception e) {
+ LOG.error("Error forwarding request to S3G endpoint", e);
+ sendErrorResponse(exchange, e);
+ } finally {
+ closeResources(conn, exchange);
+ }
+ }
+
+ private void sendErrorResponse(HttpExchange exchange, Exception e) {
+ try {
+ exchange.sendResponseHeaders(502, 0);
+ try (OutputStream os = exchange.getResponseBody()) {
+ os.write(("Proxy error: " + e.getMessage()).getBytes(StandardCharsets.UTF_8));
+ }
+ } catch (IOException ioException) {
+ LOG.error("Failed to send error response", ioException);
+ }
+ }
+
+ private void closeResources(HttpURLConnection conn, HttpExchange exchange) {
+ if (conn != null) {
+ conn.disconnect();
+ }
+ exchange.close();
+ }
+
+ private void addForwardedHeaders(HttpExchange exchange, HttpURLConnection conn) {
+ String remoteAddr = exchange.getRemoteAddress().getAddress().getHostAddress();
+ String xff = exchange.getRequestHeaders().getFirst("X-Forwarded-For");
+
+ if (StringUtils.isNotBlank(xff)) {
+ conn.setRequestProperty("X-Forwarded-For", xff + ", " + remoteAddr);
+ } else {
+ conn.setRequestProperty("X-Forwarded-For", remoteAddr);
+ }
+
+ conn.setRequestProperty("X-Forwarded-Proto", "http");
+ }
+
+ private void copyResponseBody(HttpExchange exchange, HttpURLConnection conn, int responseCode) throws IOException {
+ String transferEncoding = conn.getHeaderField("Transfer-Encoding");
+ String contentLengthStr = conn.getHeaderField("Content-Length");
+
+ if (transferEncoding != null && transferEncoding.equalsIgnoreCase("chunked")) {
+ exchange.sendResponseHeaders(responseCode, 0);
+ } else {
+ long contentLength = Optional.ofNullable(contentLengthStr).map(Long::parseLong).orElse(0L);
+ exchange.sendResponseHeaders(responseCode, contentLength);
+ }
+
+ try (InputStream is = responseCode >= 400 ? conn.getErrorStream() : conn.getInputStream();
+ OutputStream os = exchange.getResponseBody()) {
+ if (is != null) {
+ IOUtils.copyLarge(is, os, IO_BUFFER.get());
+ os.flush();
+ }
+ }
+ }
+
+ private HttpURLConnection createConnection(HttpExchange exchange, String target) throws IOException {
+ HttpURLConnection conn = (HttpURLConnection) new URL(target).openConnection();
+ String requestMethod = exchange.getRequestMethod();
+ conn.setRequestMethod(requestMethod);
+ conn.setConnectTimeout(CONNECT_TIMEOUT_MS);
+ conn.setReadTimeout(READ_TIMEOUT_MS);
+ return conn;
+ }
+
+ private void copyRequestBody(HttpExchange exchange, HttpURLConnection conn) throws IOException {
+ String method = exchange.getRequestMethod().toLowerCase();
+ if (method.equals("post") || method.equals("put") || method.equals("patch") || method.equals("delete")) {
+ String contentLength = exchange.getRequestHeaders().getFirst("Content-Length");
+ conn.setDoOutput(true);
+ if (StringUtils.isNotBlank(contentLength)) {
+ conn.setFixedLengthStreamingMode(Long.parseLong(contentLength));
+ } else {
+ conn.setChunkedStreamingMode(0);
+ }
+ try (InputStream reqBody = exchange.getRequestBody();
+ OutputStream os = conn.getOutputStream()) {
+ IOUtils.copy(reqBody, os);
+ }
+ }
+ }
+
+ private void copyRequestHeaders(HttpExchange exchange, HttpURLConnection conn) {
+ for (String headerName : exchange.getRequestHeaders().keySet()) {
+ if (isHopByHopHeader(headerName)) {
+ continue;
+ }
+
+ // Skip Expect header to avoid 100-continue issue
+ if ("Expect".equalsIgnoreCase(headerName)) {
+ continue;
+ }
Review Comment:
This issue happens when using ```HttpUrlConnection```.
Although the Jetty server does handle the ```100-continue``` response automatically, as shown [here](https://github.com/jetty/jetty.project/blob/e3fa9466633db6bf36e0eb0d17e3de166c788ede/jetty-server/src/main/java/org/eclipse/jetty/server/HttpChannelOverHttp.java#L203), it is only triggered when the input stream is read.
In some scenarios, an exception occurs before the input stream is read (for example, [here](https://github.com/apache/ozone/blob/master/hadoop-ozone/s3gateway/src/main/java/org/apache/hadoop/ozone/s3/endpoint/ObjectEndpoint.java#L243)).
This causes [HttpUrlConnection](https://github.com/AdoptOpenJDK/openjdk-jdk11/blob/master/src/java.base/share/classes/sun/net/www/protocol/http/HttpURLConnection.java#L1290) to throw a ```ProtocolException``` when it receives an unexpected HTTP response code.
This is why I decided to skip the ```Expect``` header in this case.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-11679. Support multiple S3Gs in MiniOzoneCluster [ozone]
Posted by "hevinhsu (via GitHub)" <gi...@apache.org>.
hevinhsu commented on code in PR #8733:
URL: https://github.com/apache/ozone/pull/8733#discussion_r2203144726
##########
hadoop-ozone/integration-test-s3/src/test/java/org/apache/hadoop/ozone/s3/ProxyServer.java:
##########
@@ -0,0 +1,251 @@
+/*
+ * 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.hadoop.ozone.s3;
+
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpHandler;
+import com.sun.net.httpserver.HttpServer;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.HttpURLConnection;
+import java.net.InetSocketAddress;
+import java.net.URL;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import org.apache.commons.io.IOUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * ProxyServer acts as a reverse proxy for S3G endpoints.
+ * It forwards requests to the selected S3G endpoint based on the load balancing strategy.
+ */
+public class ProxyServer {
+ private static final Logger LOG = LoggerFactory.getLogger(ProxyServer.class);
+ private final List<String> s3gEndpoints;
+ private final LoadBalanceStrategy loadBalanceStrategy;
+ private final HttpServer server;
+ private final String host;
+ private final int port;
+ private final ExecutorService executor;
+
+ // Add timeout configurations for better handling of large file operations
+ private static final int CONNECT_TIMEOUT_MS = 60000; // 60 seconds
+ private static final int READ_TIMEOUT_MS = 300000; // 5 minutes
+
+ private static final int BUFFER_SIZE = 64 * 1024; // 64KB
+ private static final ThreadLocal<byte[]> IO_BUFFER = ThreadLocal.withInitial(() -> new byte[BUFFER_SIZE]);
+
+ public ProxyServer(List<String> s3gEndpoints, String host, int proxyPort) throws Exception {
+ this(s3gEndpoints, host, proxyPort, new RoundRobinStrategy());
+ }
+
+ public ProxyServer(List<String> s3gEndpoints, String host, int proxyPort,
+ LoadBalanceStrategy loadBalanceStrategy) throws Exception {
+ this.s3gEndpoints = s3gEndpoints;
+ this.loadBalanceStrategy = loadBalanceStrategy;
+ this.host = host;
+ this.port = proxyPort;
+ server = HttpServer.create(new InetSocketAddress(host, proxyPort), 0);
+ server.createContext("/", new ProxyHandler());
+
+ this.executor = Executors.newCachedThreadPool();
+ server.setExecutor(executor);
+
+ LOG.info("ProxyServer initialized with endpoints: {}", s3gEndpoints);
+ LOG.info("Load balance strategy: {}", loadBalanceStrategy.getClass().getSimpleName());
+ LOG.info("Timeout settings - Connect: {}ms, Read: {}ms", CONNECT_TIMEOUT_MS, READ_TIMEOUT_MS);
+ }
+
+ public void start() {
+ server.start();
+ LOG.info("Proxy started on http://{}:{}", host, port);
+
+ }
+
+ public void stop() {
+ server.stop(0);
+ executor.shutdownNow();
+ LOG.info("Proxy stopped on http://{}:{}", host, port);
+ }
+
+ private class ProxyHandler implements HttpHandler {
+ @Override
+ public void handle(HttpExchange exchange) throws IOException {
+ HttpURLConnection conn = null;
+ try {
+ String target = loadBalanceStrategy.selectEndpoint(s3gEndpoints) + exchange.getRequestURI().toString();
+ conn = createConnection(exchange, target);
+ String requestMethod = conn.getRequestMethod();
+
+ LOG.info("Received request and forwarding request to [{}] {}", requestMethod, target);
+
+ copyRequestHeaders(exchange, conn);
+ addForwardedHeaders(exchange, conn);
+ copyRequestBody(exchange, conn);
+
+ int responseCode = conn.getResponseCode();
+ copyResponseHeaders(exchange, conn);
+
+ if (requestMethod.equals("HEAD") || responseCode == HttpURLConnection.HTTP_NO_CONTENT) {
+ exchange.sendResponseHeaders(responseCode, 0);
+ return;
+ }
+
+ copyResponseBody(exchange, conn, responseCode);
+
+ } catch (Exception e) {
+ LOG.error("Error forwarding request to S3G endpoint", e);
+ sendErrorResponse(exchange, e);
+ } finally {
+ closeResources(conn, exchange);
+ }
+ }
+
+ private void sendErrorResponse(HttpExchange exchange, Exception e) {
+ try {
+ exchange.sendResponseHeaders(502, 0);
+ try (OutputStream os = exchange.getResponseBody()) {
+ os.write(("Proxy error: " + e.getMessage()).getBytes(StandardCharsets.UTF_8));
+ }
+ } catch (IOException ioException) {
+ LOG.error("Failed to send error response", ioException);
+ }
+ }
+
+ private void closeResources(HttpURLConnection conn, HttpExchange exchange) {
+ if (conn != null) {
+ conn.disconnect();
+ }
+ exchange.close();
+ }
+
+ private void addForwardedHeaders(HttpExchange exchange, HttpURLConnection conn) {
+ String remoteAddr = exchange.getRemoteAddress().getAddress().getHostAddress();
+ String xff = exchange.getRequestHeaders().getFirst("X-Forwarded-For");
+
+ if (StringUtils.isNotBlank(xff)) {
+ conn.setRequestProperty("X-Forwarded-For", xff + ", " + remoteAddr);
+ } else {
+ conn.setRequestProperty("X-Forwarded-For", remoteAddr);
+ }
+
+ conn.setRequestProperty("X-Forwarded-Proto", "http");
+ }
+
+ private void copyResponseBody(HttpExchange exchange, HttpURLConnection conn, int responseCode) throws IOException {
+ String transferEncoding = conn.getHeaderField("Transfer-Encoding");
+ String contentLengthStr = conn.getHeaderField("Content-Length");
+
+ if (transferEncoding != null && transferEncoding.equalsIgnoreCase("chunked")) {
+ exchange.sendResponseHeaders(responseCode, 0);
+ } else {
+ long contentLength = Optional.ofNullable(contentLengthStr).map(Long::parseLong).orElse(0L);
+ exchange.sendResponseHeaders(responseCode, contentLength);
+ }
+
+ try (InputStream is = responseCode >= 400 ? conn.getErrorStream() : conn.getInputStream();
+ OutputStream os = exchange.getResponseBody()) {
+ if (is != null) {
+ IOUtils.copyLarge(is, os, IO_BUFFER.get());
+ os.flush();
+ }
+ }
+ }
+
+ private HttpURLConnection createConnection(HttpExchange exchange, String target) throws IOException {
+ HttpURLConnection conn = (HttpURLConnection) new URL(target).openConnection();
+ String requestMethod = exchange.getRequestMethod();
+ conn.setRequestMethod(requestMethod);
+ conn.setConnectTimeout(CONNECT_TIMEOUT_MS);
+ conn.setReadTimeout(READ_TIMEOUT_MS);
+ return conn;
+ }
+
+ private void copyRequestBody(HttpExchange exchange, HttpURLConnection conn) throws IOException {
+ String method = exchange.getRequestMethod().toLowerCase();
+ if (method.equals("post") || method.equals("put") || method.equals("patch") || method.equals("delete")) {
+ String contentLength = exchange.getRequestHeaders().getFirst("Content-Length");
+ conn.setDoOutput(true);
+ if (StringUtils.isNotBlank(contentLength)) {
+ conn.setFixedLengthStreamingMode(Long.parseLong(contentLength));
+ } else {
+ conn.setChunkedStreamingMode(0);
+ }
+ try (InputStream reqBody = exchange.getRequestBody();
+ OutputStream os = conn.getOutputStream()) {
+ IOUtils.copy(reqBody, os);
+ }
+ }
+ }
+
+ private void copyRequestHeaders(HttpExchange exchange, HttpURLConnection conn) {
+ for (String headerName : exchange.getRequestHeaders().keySet()) {
+ if (isHopByHopHeader(headerName)) {
+ continue;
+ }
+
+ // Skip Expect header to avoid 100-continue issue
+ if ("Expect".equalsIgnoreCase(headerName)) {
+ continue;
+ }
+
+ List<String> headerValues = exchange.getRequestHeaders().get(headerName);
+ if (headerValues == null || headerValues.isEmpty()) {
+ continue;
+ }
+
+ for (String value : headerValues) {
+ // Convert header names to lowercase because HttpServer capitalizes the first letter,
+ // but s3g expects lowercase header names.
+ conn.addRequestProperty(headerName.toLowerCase(), value);
+ }
Review Comment:
@ivandika3 I've opened a JIRA issue to track this: https://issues.apache.org/jira/browse/HDDS-13429
Please feel free to make any edits if needed, or let me know and I’ll update it accordingly.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-11679. Support multiple S3Gs in MiniOzoneCluster [ozone]
Posted by "ivandika3 (via GitHub)" <gi...@apache.org>.
ivandika3 commented on code in PR #8733:
URL: https://github.com/apache/ozone/pull/8733#discussion_r2202684980
##########
hadoop-ozone/integration-test-s3/src/test/java/org/apache/hadoop/ozone/s3/ProxyServerIntegrationTest.java:
##########
@@ -0,0 +1,144 @@
+/*
+ * 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.hadoop.ozone.s3;
+
+import static org.apache.ozone.test.GenericTestUtils.PortAllocator.localhostWithFreePort;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.HttpURLConnection;
+import java.net.URL;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Scanner;
+import javax.servlet.http.HttpServlet;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+import org.eclipse.jetty.server.Server;
+import org.eclipse.jetty.servlet.ServletContextHandler;
+import org.eclipse.jetty.servlet.ServletHolder;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Integration test class to verify the routing and direct access functionality
+ * of the ProxyServer.
+ */
+public class ProxyServerIntegrationTest {
+ private static final int NUM_SERVERS = 3;
+ private static final String SERVICE_PATH = "/service-name";
+
+ private static final List<Server> SERVERS = new ArrayList<>();
+ private static final List<String> ENDPOINTS = new ArrayList<>();
+ private static ProxyServer proxy;
+ private static String proxyUrl;
+
+ @BeforeAll
+ public static void setup() throws Exception {
+ for (int i = 0; i < NUM_SERVERS; i++) {
+ startMockServer("server-" + i);
+ }
+ Thread.sleep(1000);
Review Comment:
Nit: Try to avoid sleep if possible, since this might cause the tests to be flaky.
I'm OK to leave it be for now.
##########
hadoop-ozone/integration-test-s3/src/test/java/org/apache/hadoop/ozone/s3/ProxyServer.java:
##########
@@ -0,0 +1,210 @@
+/*
+ * 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.hadoop.ozone.s3;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Enumeration;
+import java.util.List;
+import javax.servlet.ServletException;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletRequestWrapper;
+import javax.servlet.http.HttpServletResponse;
+import org.eclipse.jetty.proxy.ProxyServlet;
+import org.eclipse.jetty.server.Server;
+import org.eclipse.jetty.servlet.ServletContextHandler;
+import org.eclipse.jetty.servlet.ServletHolder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * ProxyServer acts as a reverse proxy for S3G endpoints.
+ * It forwards requests to the selected S3G endpoint based on the load balancing strategy.
+ */
+public class ProxyServer {
+ private static final Logger LOG = LoggerFactory.getLogger(ProxyServer.class);
+ private final List<String> s3gEndpoints;
+ private final LoadBalanceStrategy loadBalanceStrategy;
+ private final Server server;
+ private final String host;
+ private final int port;
+
+ public ProxyServer(List<String> s3gEndpoints, String host, int proxyPort) throws Exception {
+ this(s3gEndpoints, host, proxyPort, new RoundRobinStrategy());
+ }
+
+ public ProxyServer(List<String> s3gEndpoints, String host, int proxyPort,
+ LoadBalanceStrategy loadBalanceStrategy) throws Exception {
+ this.s3gEndpoints = s3gEndpoints;
+ this.loadBalanceStrategy = loadBalanceStrategy;
+ this.host = host;
+ this.port = proxyPort;
+
+ server = new Server(proxyPort);
+ ServletContextHandler context = new ServletContextHandler();
+ context.setContextPath("/");
+
+ ProxyHandler proxyHandler = new ProxyHandler();
+ ServletHolder proxyHolder = new ServletHolder(proxyHandler);
+
+ proxyHolder.setInitParameter("proxyTo", "");
+
+
+ context.addServlet(proxyHolder, "/*");
+ server.setHandler(context);
+
+ LOG.info("ProxyServer initialized with endpoints: {}", s3gEndpoints);
+ LOG.info("Load balance strategy: {}", loadBalanceStrategy.getClass().getSimpleName());
+ }
+
+ public void start() throws Exception {
+ server.start();
+ LOG.info("Proxy started on http://{}:{}", host, port);
+ }
+
+ public void stop() throws Exception {
+ server.stop();
+ LOG.info("Proxy stopped on http://{}:{}", host, port);
+ }
+
+ /**
+ * ProxyHandler is a subclass of Jetty's ProxyServlet.Transparent.
+ * It implements logic for request rewriting, service handling,
+ * proxy response failure, and rewrite failure handling.
+ * This handler is mainly used to forward requests to different S3G endpoints
+ * based on the load balancing strategy, handle special HTTP headers (such as Expect),
+ * and manage exceptions during the proxy process.
+ */
+ public class ProxyHandler extends ProxyServlet.Transparent {
+
+ @Override
+ public void init() throws ServletException {
+ super.init();
+
+ LOG.info("MyProxyHandler initialized with {} endpoints",
+ s3gEndpoints != null ? s3gEndpoints.size() : 0);
+ if (s3gEndpoints != null) {
+ LOG.info("Endpoints: {}", s3gEndpoints);
+ }
+ LOG.info("Load balance strategy: {}", loadBalanceStrategy.getClass().getSimpleName());
+ }
+
+ @Override
+ protected String rewriteTarget(HttpServletRequest request) {
+
+ String baseUrl = loadBalanceStrategy.selectEndpoint(s3gEndpoints);
+
+ String requestUri = request.getRequestURI();
+ String queryString = request.getQueryString();
+
+ StringBuilder targetUrl = new StringBuilder(baseUrl);
+ if (requestUri != null) {
+ targetUrl.append(requestUri);
+ }
+ if (queryString != null) {
+ targetUrl.append('?').append(queryString);
+ }
+
+ String finalUrl = targetUrl.toString();
+ LOG.info("Rewriting target URL: [{}] {}", request.getMethod(), finalUrl);
+ return finalUrl;
+ }
+
+ @Override
+ protected void service(HttpServletRequest request, HttpServletResponse response)
+ throws ServletException, IOException {
+
+ // Remove the Expect header to avoid Jetty's 100-continue handling issue:
+ // In some scenarios (e.g. testPutObjectEmpty), when content-length != 0, Jetty triggers the 100-continue process,
+ // causing the server to wait for a 100 response code. However, Jetty only sends the 100 response
+ // when the InputStream is read, which can lead to request failures.
+ if (request.getHeader("Expect") != null) {
Review Comment:
Nit: Replace all the `Expect` with a constant (the http dependencies should have one).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-11679. Support multiple S3Gs in MiniOzoneCluster [ozone]
Posted by "hevinhsu (via GitHub)" <gi...@apache.org>.
hevinhsu commented on code in PR #8733:
URL: https://github.com/apache/ozone/pull/8733#discussion_r2186990992
##########
hadoop-ozone/integration-test-s3/src/test/java/org/apache/hadoop/ozone/s3/ProxyServer.java:
##########
@@ -0,0 +1,251 @@
+/*
+ * 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.hadoop.ozone.s3;
+
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpHandler;
+import com.sun.net.httpserver.HttpServer;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.HttpURLConnection;
+import java.net.InetSocketAddress;
+import java.net.URL;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import org.apache.commons.io.IOUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * ProxyServer acts as a reverse proxy for S3G endpoints.
+ * It forwards requests to the selected S3G endpoint based on the load balancing strategy.
+ */
+public class ProxyServer {
+ private static final Logger LOG = LoggerFactory.getLogger(ProxyServer.class);
+ private final List<String> s3gEndpoints;
+ private final LoadBalanceStrategy loadBalanceStrategy;
+ private final HttpServer server;
+ private final String host;
+ private final int port;
+ private final ExecutorService executor;
+
+ // Add timeout configurations for better handling of large file operations
+ private static final int CONNECT_TIMEOUT_MS = 60000; // 60 seconds
+ private static final int READ_TIMEOUT_MS = 300000; // 5 minutes
+
+ private static final int BUFFER_SIZE = 64 * 1024; // 64KB
+ private static final ThreadLocal<byte[]> IO_BUFFER = ThreadLocal.withInitial(() -> new byte[BUFFER_SIZE]);
+
+ public ProxyServer(List<String> s3gEndpoints, String host, int proxyPort) throws Exception {
+ this(s3gEndpoints, host, proxyPort, new RoundRobinStrategy());
+ }
+
+ public ProxyServer(List<String> s3gEndpoints, String host, int proxyPort,
+ LoadBalanceStrategy loadBalanceStrategy) throws Exception {
+ this.s3gEndpoints = s3gEndpoints;
+ this.loadBalanceStrategy = loadBalanceStrategy;
+ this.host = host;
+ this.port = proxyPort;
+ server = HttpServer.create(new InetSocketAddress(host, proxyPort), 0);
+ server.createContext("/", new ProxyHandler());
+
+ this.executor = Executors.newCachedThreadPool();
+ server.setExecutor(executor);
+
+ LOG.info("ProxyServer initialized with endpoints: {}", s3gEndpoints);
+ LOG.info("Load balance strategy: {}", loadBalanceStrategy.getClass().getSimpleName());
+ LOG.info("Timeout settings - Connect: {}ms, Read: {}ms", CONNECT_TIMEOUT_MS, READ_TIMEOUT_MS);
+ }
+
+ public void start() {
+ server.start();
+ LOG.info("Proxy started on http://{}:{}", host, port);
+
+ }
+
+ public void stop() {
+ server.stop(0);
+ executor.shutdownNow();
+ LOG.info("Proxy stopped on http://{}:{}", host, port);
+ }
+
+ private class ProxyHandler implements HttpHandler {
+ @Override
+ public void handle(HttpExchange exchange) throws IOException {
+ HttpURLConnection conn = null;
+ try {
+ String target = loadBalanceStrategy.selectEndpoint(s3gEndpoints) + exchange.getRequestURI().toString();
+ conn = createConnection(exchange, target);
+ String requestMethod = conn.getRequestMethod();
+
+ LOG.info("Received request and forwarding request to [{}] {}", requestMethod, target);
+
+ copyRequestHeaders(exchange, conn);
+ addForwardedHeaders(exchange, conn);
+ copyRequestBody(exchange, conn);
+
+ int responseCode = conn.getResponseCode();
+ copyResponseHeaders(exchange, conn);
+
+ if (requestMethod.equals("HEAD") || responseCode == HttpURLConnection.HTTP_NO_CONTENT) {
+ exchange.sendResponseHeaders(responseCode, 0);
+ return;
+ }
+
+ copyResponseBody(exchange, conn, responseCode);
+
+ } catch (Exception e) {
+ LOG.error("Error forwarding request to S3G endpoint", e);
+ sendErrorResponse(exchange, e);
+ } finally {
+ closeResources(conn, exchange);
+ }
+ }
+
+ private void sendErrorResponse(HttpExchange exchange, Exception e) {
+ try {
+ exchange.sendResponseHeaders(502, 0);
+ try (OutputStream os = exchange.getResponseBody()) {
+ os.write(("Proxy error: " + e.getMessage()).getBytes(StandardCharsets.UTF_8));
+ }
+ } catch (IOException ioException) {
+ LOG.error("Failed to send error response", ioException);
+ }
+ }
+
+ private void closeResources(HttpURLConnection conn, HttpExchange exchange) {
+ if (conn != null) {
+ conn.disconnect();
+ }
+ exchange.close();
+ }
+
+ private void addForwardedHeaders(HttpExchange exchange, HttpURLConnection conn) {
+ String remoteAddr = exchange.getRemoteAddress().getAddress().getHostAddress();
+ String xff = exchange.getRequestHeaders().getFirst("X-Forwarded-For");
+
+ if (StringUtils.isNotBlank(xff)) {
+ conn.setRequestProperty("X-Forwarded-For", xff + ", " + remoteAddr);
+ } else {
+ conn.setRequestProperty("X-Forwarded-For", remoteAddr);
+ }
+
+ conn.setRequestProperty("X-Forwarded-Proto", "http");
+ }
+
+ private void copyResponseBody(HttpExchange exchange, HttpURLConnection conn, int responseCode) throws IOException {
+ String transferEncoding = conn.getHeaderField("Transfer-Encoding");
+ String contentLengthStr = conn.getHeaderField("Content-Length");
+
+ if (transferEncoding != null && transferEncoding.equalsIgnoreCase("chunked")) {
+ exchange.sendResponseHeaders(responseCode, 0);
+ } else {
+ long contentLength = Optional.ofNullable(contentLengthStr).map(Long::parseLong).orElse(0L);
+ exchange.sendResponseHeaders(responseCode, contentLength);
+ }
+
+ try (InputStream is = responseCode >= 400 ? conn.getErrorStream() : conn.getInputStream();
+ OutputStream os = exchange.getResponseBody()) {
+ if (is != null) {
+ IOUtils.copyLarge(is, os, IO_BUFFER.get());
+ os.flush();
+ }
+ }
+ }
+
+ private HttpURLConnection createConnection(HttpExchange exchange, String target) throws IOException {
+ HttpURLConnection conn = (HttpURLConnection) new URL(target).openConnection();
+ String requestMethod = exchange.getRequestMethod();
+ conn.setRequestMethod(requestMethod);
+ conn.setConnectTimeout(CONNECT_TIMEOUT_MS);
+ conn.setReadTimeout(READ_TIMEOUT_MS);
+ return conn;
+ }
+
+ private void copyRequestBody(HttpExchange exchange, HttpURLConnection conn) throws IOException {
+ String method = exchange.getRequestMethod().toLowerCase();
+ if (method.equals("post") || method.equals("put") || method.equals("patch") || method.equals("delete")) {
+ String contentLength = exchange.getRequestHeaders().getFirst("Content-Length");
+ conn.setDoOutput(true);
+ if (StringUtils.isNotBlank(contentLength)) {
+ conn.setFixedLengthStreamingMode(Long.parseLong(contentLength));
+ } else {
+ conn.setChunkedStreamingMode(0);
+ }
+ try (InputStream reqBody = exchange.getRequestBody();
+ OutputStream os = conn.getOutputStream()) {
+ IOUtils.copy(reqBody, os);
+ }
+ }
+ }
+
+ private void copyRequestHeaders(HttpExchange exchange, HttpURLConnection conn) {
+ for (String headerName : exchange.getRequestHeaders().keySet()) {
+ if (isHopByHopHeader(headerName)) {
+ continue;
+ }
+
+ // Skip Expect header to avoid 100-continue issue
+ if ("Expect".equalsIgnoreCase(headerName)) {
+ continue;
+ }
+
+ List<String> headerValues = exchange.getRequestHeaders().get(headerName);
+ if (headerValues == null || headerValues.isEmpty()) {
+ continue;
+ }
+
+ for (String value : headerValues) {
+ // Convert header names to lowercase because HttpServer capitalizes the first letter,
+ // but s3g expects lowercase header names.
+ conn.addRequestProperty(headerName.toLowerCase(), value);
+ }
Review Comment:
Yes, you’re right.
I converted the headers to lower case because the ```testLowLevelMultipartUpload``` test was failing ([link](https://github.com/apache/ozone/blob/master/hadoop-ozone/integration-test-s3/src/test/java/org/apache/hadoop/ozone/s3/awssdk/v1/AbstractS3SDKV1Tests.java#L730)).
This happens because the current implementation compares HTTP header prefixes without ignoring case sensitivity ([link](https://github.com/apache/ozone/blob/master/hadoop-ozone/s3gateway/src/main/java/org/apache/hadoop/ozone/s3/endpoint/EndpointBase.java#L298-L308)).
I think I should remove the .toLowerCase() here and create a separate issue to fix this API in S3G.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-11679. Support multiple S3Gs in MiniOzoneCluster [ozone]
Posted by "ivandika3 (via GitHub)" <gi...@apache.org>.
ivandika3 commented on PR #8733:
URL: https://github.com/apache/ozone/pull/8733#issuecomment-3038837790
> Regarding the unit test, my initial thought is to add a unit test for RoundRobinStrategy, but I’m not sure if that would be sufficient.
Yes, we can add a simple unit test on the `RoundRobinStrategy`, although it should be straightforward. The more important test is to validate that the proxy actual routes to different S3Gs, maybe through using `Spy` by parsing the S3G audit logs.
> As for merging with the S3G integration tests, do you mean copying the existing integration test classes and replacing S3GatewayService with MultiS3GatewayService?
We incorporate the changes in your fork (https://github.com/hevinhsu/ozone/actions/runs/16043512252). I'm OK with replacing the `S3GatewayService` to `MultiS3GatewayService` as long as the tests are fine.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-11679. Support multiple S3Gs in MiniOzoneCluster [ozone]
Posted by "ivandika3 (via GitHub)" <gi...@apache.org>.
ivandika3 commented on code in PR #8733:
URL: https://github.com/apache/ozone/pull/8733#discussion_r2187147506
##########
hadoop-ozone/integration-test-s3/src/test/java/org/apache/hadoop/ozone/s3/ProxyServer.java:
##########
@@ -0,0 +1,251 @@
+/*
+ * 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.hadoop.ozone.s3;
+
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpHandler;
+import com.sun.net.httpserver.HttpServer;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.HttpURLConnection;
+import java.net.InetSocketAddress;
+import java.net.URL;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import org.apache.commons.io.IOUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * ProxyServer acts as a reverse proxy for S3G endpoints.
+ * It forwards requests to the selected S3G endpoint based on the load balancing strategy.
+ */
+public class ProxyServer {
+ private static final Logger LOG = LoggerFactory.getLogger(ProxyServer.class);
+ private final List<String> s3gEndpoints;
+ private final LoadBalanceStrategy loadBalanceStrategy;
+ private final HttpServer server;
+ private final String host;
+ private final int port;
+ private final ExecutorService executor;
+
+ // Add timeout configurations for better handling of large file operations
+ private static final int CONNECT_TIMEOUT_MS = 60000; // 60 seconds
+ private static final int READ_TIMEOUT_MS = 300000; // 5 minutes
+
+ private static final int BUFFER_SIZE = 64 * 1024; // 64KB
+ private static final ThreadLocal<byte[]> IO_BUFFER = ThreadLocal.withInitial(() -> new byte[BUFFER_SIZE]);
+
+ public ProxyServer(List<String> s3gEndpoints, String host, int proxyPort) throws Exception {
+ this(s3gEndpoints, host, proxyPort, new RoundRobinStrategy());
+ }
+
+ public ProxyServer(List<String> s3gEndpoints, String host, int proxyPort,
+ LoadBalanceStrategy loadBalanceStrategy) throws Exception {
+ this.s3gEndpoints = s3gEndpoints;
+ this.loadBalanceStrategy = loadBalanceStrategy;
+ this.host = host;
+ this.port = proxyPort;
+ server = HttpServer.create(new InetSocketAddress(host, proxyPort), 0);
+ server.createContext("/", new ProxyHandler());
+
+ this.executor = Executors.newCachedThreadPool();
+ server.setExecutor(executor);
+
+ LOG.info("ProxyServer initialized with endpoints: {}", s3gEndpoints);
+ LOG.info("Load balance strategy: {}", loadBalanceStrategy.getClass().getSimpleName());
+ LOG.info("Timeout settings - Connect: {}ms, Read: {}ms", CONNECT_TIMEOUT_MS, READ_TIMEOUT_MS);
+ }
+
+ public void start() {
+ server.start();
+ LOG.info("Proxy started on http://{}:{}", host, port);
+
+ }
+
+ public void stop() {
+ server.stop(0);
+ executor.shutdownNow();
+ LOG.info("Proxy stopped on http://{}:{}", host, port);
+ }
+
+ private class ProxyHandler implements HttpHandler {
+ @Override
+ public void handle(HttpExchange exchange) throws IOException {
+ HttpURLConnection conn = null;
+ try {
+ String target = loadBalanceStrategy.selectEndpoint(s3gEndpoints) + exchange.getRequestURI().toString();
+ conn = createConnection(exchange, target);
+ String requestMethod = conn.getRequestMethod();
+
+ LOG.info("Received request and forwarding request to [{}] {}", requestMethod, target);
+
+ copyRequestHeaders(exchange, conn);
+ addForwardedHeaders(exchange, conn);
+ copyRequestBody(exchange, conn);
+
+ int responseCode = conn.getResponseCode();
+ copyResponseHeaders(exchange, conn);
+
+ if (requestMethod.equals("HEAD") || responseCode == HttpURLConnection.HTTP_NO_CONTENT) {
+ exchange.sendResponseHeaders(responseCode, 0);
+ return;
+ }
+
+ copyResponseBody(exchange, conn, responseCode);
+
+ } catch (Exception e) {
+ LOG.error("Error forwarding request to S3G endpoint", e);
+ sendErrorResponse(exchange, e);
+ } finally {
+ closeResources(conn, exchange);
+ }
+ }
+
+ private void sendErrorResponse(HttpExchange exchange, Exception e) {
+ try {
+ exchange.sendResponseHeaders(502, 0);
+ try (OutputStream os = exchange.getResponseBody()) {
+ os.write(("Proxy error: " + e.getMessage()).getBytes(StandardCharsets.UTF_8));
+ }
+ } catch (IOException ioException) {
+ LOG.error("Failed to send error response", ioException);
+ }
+ }
+
+ private void closeResources(HttpURLConnection conn, HttpExchange exchange) {
+ if (conn != null) {
+ conn.disconnect();
+ }
+ exchange.close();
+ }
+
+ private void addForwardedHeaders(HttpExchange exchange, HttpURLConnection conn) {
+ String remoteAddr = exchange.getRemoteAddress().getAddress().getHostAddress();
+ String xff = exchange.getRequestHeaders().getFirst("X-Forwarded-For");
+
+ if (StringUtils.isNotBlank(xff)) {
+ conn.setRequestProperty("X-Forwarded-For", xff + ", " + remoteAddr);
+ } else {
+ conn.setRequestProperty("X-Forwarded-For", remoteAddr);
+ }
+
+ conn.setRequestProperty("X-Forwarded-Proto", "http");
+ }
+
+ private void copyResponseBody(HttpExchange exchange, HttpURLConnection conn, int responseCode) throws IOException {
+ String transferEncoding = conn.getHeaderField("Transfer-Encoding");
+ String contentLengthStr = conn.getHeaderField("Content-Length");
+
+ if (transferEncoding != null && transferEncoding.equalsIgnoreCase("chunked")) {
+ exchange.sendResponseHeaders(responseCode, 0);
+ } else {
+ long contentLength = Optional.ofNullable(contentLengthStr).map(Long::parseLong).orElse(0L);
+ exchange.sendResponseHeaders(responseCode, contentLength);
+ }
+
+ try (InputStream is = responseCode >= 400 ? conn.getErrorStream() : conn.getInputStream();
+ OutputStream os = exchange.getResponseBody()) {
+ if (is != null) {
+ IOUtils.copyLarge(is, os, IO_BUFFER.get());
+ os.flush();
+ }
+ }
+ }
+
+ private HttpURLConnection createConnection(HttpExchange exchange, String target) throws IOException {
+ HttpURLConnection conn = (HttpURLConnection) new URL(target).openConnection();
+ String requestMethod = exchange.getRequestMethod();
+ conn.setRequestMethod(requestMethod);
+ conn.setConnectTimeout(CONNECT_TIMEOUT_MS);
+ conn.setReadTimeout(READ_TIMEOUT_MS);
+ return conn;
+ }
+
+ private void copyRequestBody(HttpExchange exchange, HttpURLConnection conn) throws IOException {
+ String method = exchange.getRequestMethod().toLowerCase();
+ if (method.equals("post") || method.equals("put") || method.equals("patch") || method.equals("delete")) {
+ String contentLength = exchange.getRequestHeaders().getFirst("Content-Length");
+ conn.setDoOutput(true);
+ if (StringUtils.isNotBlank(contentLength)) {
+ conn.setFixedLengthStreamingMode(Long.parseLong(contentLength));
+ } else {
+ conn.setChunkedStreamingMode(0);
+ }
+ try (InputStream reqBody = exchange.getRequestBody();
+ OutputStream os = conn.getOutputStream()) {
+ IOUtils.copy(reqBody, os);
+ }
+ }
+ }
+
+ private void copyRequestHeaders(HttpExchange exchange, HttpURLConnection conn) {
+ for (String headerName : exchange.getRequestHeaders().keySet()) {
+ if (isHopByHopHeader(headerName)) {
+ continue;
+ }
+
+ // Skip Expect header to avoid 100-continue issue
+ if ("Expect".equalsIgnoreCase(headerName)) {
+ continue;
+ }
+
+ List<String> headerValues = exchange.getRequestHeaders().get(headerName);
+ if (headerValues == null || headerValues.isEmpty()) {
+ continue;
+ }
+
+ for (String value : headerValues) {
+ // Convert header names to lowercase because HttpServer capitalizes the first letter,
+ // but s3g expects lowercase header names.
+ conn.addRequestProperty(headerName.toLowerCase(), value);
+ }
Review Comment:
> I think I should remove the .toLowerCase() here and create a separate issue to fix this API in S3G.
Sounds good.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org