feat: log non-2xx responses in the HTTP data plane source

CHANGELOG.md

@@ -6,6 +6,8 @@
 
 #### Changes
 
+- `HttpDataSource` now logs a warning when the data source returns a non-2xx status code
+
 #### Details
 
 #### Compatibility

extensions/data-plane/data-plane-http/src/main/java/org/eclipse/edc/connector/dataplane/http/pipeline/HttpDataSource.java

@@ -16,6 +16,7 @@
 
 
 import okhttp3.MediaType;
+import okhttp3.Response;
 import okhttp3.ResponseBody;
 import org.eclipse.edc.connector.dataplane.http.params.HttpRequestFactory;
 import org.eclipse.edc.connector.dataplane.http.spi.HttpRequestParams;
@@ -38,9 +39,10 @@
 import static org.eclipse.edc.connector.dataplane.spi.pipeline.StreamResult.success;
 
 public class HttpDataSource implements DataSource {
-    private static final int FORBIDDEN = 401;
-    private static final int NOT_AUTHORIZED = 403;
+    private static final int UNAUTHORIZED = 401;
+    private static final int FORBIDDEN = 403;
     private static final int NOT_FOUND = 404;
+    private static final long ERROR_BODY_LOG_LIMIT = 4096;
 
     private String name;
     private HttpRequestParams params;
@@ -71,7 +73,10 @@ public StreamResult<Stream<Part>> openPartStream() {
                 return success(Stream.of(new HttpPart(name, stream, mediaType)));
             } else {
                 try {
-                    if (NOT_AUTHORIZED == response.code() || FORBIDDEN == response.code()) {
+                    monitor.warning(format("Error {%s: %s} received reading HTTP data from endpoint '%s %s'. Response body: '%s'",
+                            response.code(), response.message(), request.method(), request.url().url(), readErrorBody(response)));
+
+                    if (UNAUTHORIZED == response.code() || FORBIDDEN == response.code()) {
                         return StreamResult.notAuthorized();
                     } else if (NOT_FOUND == response.code()) {
                         return StreamResult.notFound();
@@ -92,6 +97,14 @@ public StreamResult<Stream<Part>> openPartStream() {
 
     }
 
+    private String readErrorBody(Response response) {
+        try {
+            return response.peekBody(ERROR_BODY_LOG_LIMIT).string();
+        } catch (Exception e) {
+            return "HTTP response error body could not be read: " + e;
+        }
+    }
+
     @Override
     public void close() {
         var bodyStream = responseBodyStream.get();

extensions/data-plane/data-plane-http/src/test/java/org/eclipse/edc/connector/dataplane/http/pipeline/HttpDataSourceTest.java

@@ -31,6 +31,7 @@
 import org.junit.jupiter.params.provider.Arguments;
 import org.junit.jupiter.params.provider.ArgumentsProvider;
 import org.junit.jupiter.params.provider.ArgumentsSource;
+import org.mockito.ArgumentCaptor;
 
 import java.io.IOException;
 import java.io.InputStream;
@@ -91,6 +92,29 @@ void verifyCallFailed(int code, StreamFailure.Reason reason) {
         verify(requestFactory).toRequest(any());
     }
 
+    @ParameterizedTest
+    @ArgumentsSource(StreamFailureArguments.class)
+    void verifyCallFailed_logsStatusCodeUrlAndResponseBody(int code, StreamFailure.Reason reason) {
+        var responseBody = ResponseBody.create("Test body", MediaType.parse("text/plain"));
+        var interceptor = new CustomInterceptor(code, responseBody, "Test message");
+        var monitor = mock(Monitor.class);
+        var source = defaultBuilder(interceptor).params(mock()).requestFactory(requestFactory).monitor(monitor).build();
+        when(requestFactory.toRequest(any())).thenReturn(dummyRequest());
+
+        var result = source.openPartStream();
+
+        var logLine = ArgumentCaptor.forClass(String.class);
+        verify(monitor).warning(logLine.capture());
+        assertThat(logLine.getValue())
+                .contains(String.valueOf(code))
+                .contains("Test message")
+                .contains("http://some.test.url/")
+                .contains("Test body");
+        // the response body is logged but deliberately not added to the failure returned to the consumer
+        assertThat(result).isFailed().extracting(StreamFailure::getReason).isEqualTo(reason);
+        assertThat(result.getFailureMessages()).noneMatch(it -> it.contains("Test body"));
+    }
+
     @Test
     void close_shouldCloseResponseBodyAndStream() throws IOException {
         InputStream stream = mock();