feat: log http response body in debug mode (#95)

* feat: log http response body when log_level equals debug

* refactor: reset input stream after logging response

* refactor: fix prom sink tests by mocking response body

* refactor: print response body via toString method

* feat: log http response body if debug is enabled

* refactor: remove serializableHttpResponse class

Co-authored-by: mayur.gubrele <2310-mayur.gubrele@users.noreply.source.golabs.io>

src/main/java/io/odpf/firehose/metrics/Instrumentation.java

@@ -75,6 +75,9 @@ public void logError(String template, Object... t) {
         logger.error(template, t);
     }
 
+    public boolean isDebugEnabled() {
+        return logger.isDebugEnabled();
+    }
     // ============== FILTER MESSAGES ==============
 
     /**

src/main/java/io/odpf/firehose/sink/common/AbstractHttpSink.java

@@ -14,12 +14,17 @@
 import org.apache.http.client.methods.HttpEntityEnclosingRequestBase;
 import org.apache.http.util.EntityUtils;
 
+import java.io.BufferedReader;
 import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
 import java.util.regex.Pattern;
+import java.util.stream.Collectors;
 
 import static io.odpf.firehose.metrics.Metrics.SINK_HTTP_RESPONSE_CODE_TOTAL;
 
@@ -49,6 +54,9 @@ public List<Message> execute() throws Exception {
                 response = httpClient.execute(httpRequest);
                 List<String> contentStringList = null;
                 getInstrumentation().logInfo("Response Status: {}", statusCode(response));
+                if (shouldLogResponse(response)) {
+                    printResponse(response);
+                }
                 if (shouldLogRequest(response)) {
                     contentStringList = readContent(httpRequest);
                     printRequest(httpRequest, contentStringList);
@@ -88,6 +96,10 @@ private boolean shouldLogRequest(HttpResponse response) {
         return response == null || getRequestLogStatusCodeRanges().containsKey(response.getStatusLine().getStatusCode());
     }
 
+    private boolean shouldLogResponse(HttpResponse response) {
+        return getInstrumentation().isDebugEnabled() && response != null;
+    }
+
     private boolean shouldRetry(HttpResponse response) {
         return response == null || getRetryStatusCodeRanges().containsKey(response.getStatusLine().getStatusCode());
     }
@@ -119,6 +131,16 @@ private void printRequest(HttpEntityEnclosingRequestBase httpRequest, List<Strin
         getInstrumentation().logInfo(entireRequest);
     }
 
+    private void printResponse(HttpResponse httpResponse) throws IOException {
+        try (InputStream inputStream = httpResponse.getEntity().getContent()) {
+            String responseBody = String.format("Response Body: %s",
+                    Strings.join(new BufferedReader(new InputStreamReader(
+                            inputStream,
+                            StandardCharsets.UTF_8)).lines().collect(Collectors.toList()), "\n"));
+            getInstrumentation().logDebug(responseBody);
+        }
+    }
+
     protected abstract List<String> readContent(HttpEntityEnclosingRequestBase httpRequest) throws IOException;
 
     protected abstract void captureMessageDropCount(HttpResponse response, List<String> contentString) throws IOException;

src/test/java/io/odpf/firehose/sink/http/HttpSinkTest.java

@@ -407,4 +407,26 @@ public void shouldCaptureResponseStatusCount() throws Exception {
 
         verify(instrumentation, times(1)).captureCountWithTags("firehose_sink_http_response_code_total", 1, "status_code=" + statusLine.getStatusCode(), "url=" + uri.getPath());
     }
+
+    @Test
+    public void shouldLogResponseBodyWhenDebugIsEnabledAndNonNullResponse() throws Exception {
+        when(response.getStatusLine()).thenReturn(statusLine);
+        when(statusLine.getStatusCode()).thenReturn(200);
+
+        List<HttpEntityEnclosingRequestBase> httpRequests = Collections.singletonList(httpPut);
+
+        when(httpPut.getMethod()).thenReturn("PUT");
+        when(httpPut.getURI()).thenReturn(new URI("http://dummy.com"));
+        when(httpClient.execute(httpPut)).thenReturn(response);
+        when(response.getEntity()).thenReturn(httpEntity);
+        when(httpEntity.getContent()).thenReturn(new StringInputStream("[{\"key\":\"value1\"},{\"key\":\"value2\"}]"));
+        when(request.build(messages)).thenReturn(httpRequests);
+        when(instrumentation.isDebugEnabled()).thenReturn(true);
+
+        HttpSink httpSink = new HttpSink(instrumentation, request, httpClient, stencilClient,
+                retryStatusCodeRange, requestLogStatusCodeRanges);
+        httpSink.prepare(messages);
+        httpSink.execute();
+        verify(instrumentation, times(1)).logDebug("Response Body: [{\"key\":\"value1\"},{\"key\":\"value2\"}]");
+    }
 }