feat: put method in http postprocessor (#217)

* feat: put method in http postprocessor * feat: add endpointVariables functionality in http GET postprocessor * chore: bump version * add: unit test to parse endpointVariables * remove: generated source * add: include generated-sources in gitignore * fix: null string validation * doc: update documentation * refactor: use []Object instead of string to handle endpointVariables * remove: empty string validation * test: add more unit test for multiple endpointVariables case
raystack · Nov 24, 2022 · 0e36063 · 0e36063
1 parent e46f634
commit 0e36063
Show file tree

Hide file tree

Showing 19 changed files with 282 additions and 106 deletions.
diff --git a/.gitignore b/.gitignore
@@ -12,4 +12,5 @@ target/
 bin
 .settings
 .gradletasknamecache
-.DS_Store
+.DS_Store
+dagger-common/src/generated-sources/
diff --git a/...r-core/src/main/java/io/odpf/dagger/core/processors/external/http/HttpAsyncConnector.java b/...r-core/src/main/java/io/odpf/dagger/core/processors/external/http/HttpAsyncConnector.java
@@ -98,11 +98,13 @@ protected void process(Row input, ResultFuture<Row> resultFuture) {
                     .getVariablesValue(rowManager, ExternalPostProcessorVariableType.REQUEST_VARIABLES, httpSourceConfig.getRequestVariables(), resultFuture);
             Object[] dynamicHeaderVariablesValues = getEndpointHandler()
                     .getVariablesValue(rowManager, ExternalPostProcessorVariableType.HEADER_VARIABLES, httpSourceConfig.getHeaderVariables(), resultFuture);
+            Object[] endpointVariablesValues = getEndpointHandler()
+                    .getVariablesValue(rowManager, ExternalPostProcessorVariableType.ENDPOINT_VARIABLE, httpSourceConfig.getEndpointVariables(), resultFuture);
             if (getEndpointHandler().isQueryInvalid(resultFuture, rowManager, httpSourceConfig.getRequestVariables(), requestVariablesValues) || getEndpointHandler().isQueryInvalid(resultFuture, rowManager, httpSourceConfig.getHeaderVariables(), dynamicHeaderVariablesValues)) {
                 return;
             }
 
-            BoundRequestBuilder request = HttpRequestFactory.createRequest(httpSourceConfig, httpClient, requestVariablesValues, dynamicHeaderVariablesValues);
+            BoundRequestBuilder request = HttpRequestFactory.createRequest(httpSourceConfig, httpClient, requestVariablesValues, dynamicHeaderVariablesValues, endpointVariablesValues);
             HttpResponseHandler httpResponseHandler = new HttpResponseHandler(httpSourceConfig, getMeterStatsManager(),
                     rowManager, getColumnNameManager(), getOutputDescriptor(resultFuture), resultFuture, getErrorReporter(), new PostResponseTelemetry());
             httpResponseHandler.startTimer();

diff --git a/dagger-core/src/main/java/io/odpf/dagger/core/processors/external/http/HttpSourceConfig.java b/dagger-core/src/main/java/io/odpf/dagger/core/processors/external/http/HttpSourceConfig.java
@@ -17,6 +17,7 @@
  */
 public class HttpSourceConfig implements Serializable, SourceConfig {
     private String endpoint;
+    private String endpointVariables;
     private String verb;
     private String requestPattern;
     private String requestVariables;
@@ -39,6 +40,7 @@ public class HttpSourceConfig implements Serializable, SourceConfig {
      * Instantiates a new Http source config.
      *
      * @param endpoint           the endpoint
+     * @param endpointVariables  the endpoint variables
      * @param verb               the verb
      * @param requestPattern     the request pattern
      * @param requestVariables   the request variables
@@ -54,8 +56,9 @@ public class HttpSourceConfig implements Serializable, SourceConfig {
      * @param metricId           the metric id
      * @param retainResponseType the retain response type
      */
-    public HttpSourceConfig(String endpoint, String verb, String requestPattern, String requestVariables, String headerPattern, String headerVariables, String streamTimeout, String connectTimeout, boolean failOnErrors, String type, String capacity, Map<String, String> headers, Map<String, OutputMapping> outputMapping, String metricId, boolean retainResponseType) {
+    public HttpSourceConfig(String endpoint, String endpointVariables, String verb, String requestPattern, String requestVariables, String headerPattern, String headerVariables, String streamTimeout, String connectTimeout, boolean failOnErrors, String type, String capacity, Map<String, String> headers, Map<String, OutputMapping> outputMapping, String metricId, boolean retainResponseType) {
         this.endpoint = endpoint;
+        this.endpointVariables = endpointVariables;
         this.verb = verb;
         this.requestPattern = requestPattern;
         this.requestVariables = requestVariables;
@@ -90,6 +93,16 @@ public String getEndpoint() {
         return endpoint;
     }
 
+    /**
+     * Gets endpoint variables.
+     *
+     * @return the endpointVariables
+     */
+    public String getEndpointVariables() {
+        return endpointVariables;
+    }
+
+
     /**
      * Gets verb.
      *
@@ -237,6 +250,6 @@ public boolean equals(Object o) {
 
     @Override
     public int hashCode() {
-        return Objects.hash(endpoint, verb, requestPattern, requestVariables, headerPattern, headerVariables, streamTimeout, connectTimeout, failOnErrors, type, capacity, headers, outputMapping, metricId, retainResponseType);
+        return Objects.hash(endpoint, endpointVariables, verb, requestPattern, requestVariables, headerPattern, headerVariables, streamTimeout, connectTimeout, failOnErrors, type, capacity, headers, outputMapping, metricId, retainResponseType);
     }
 }
diff --git a/...main/java/io/odpf/dagger/core/processors/external/http/request/HttpGetRequestHandler.java b/...main/java/io/odpf/dagger/core/processors/external/http/request/HttpGetRequestHandler.java
@@ -19,25 +19,29 @@ public class HttpGetRequestHandler implements HttpRequestHandler {
     private AsyncHttpClient httpClient;
     private Object[] requestVariablesValues;
     private Object[] dynamicHeaderVariablesValues;
+    private Object[] endpointVariablesValues;
 
     /**
      * Instantiates a new Http get request handler.
      *
-     * @param httpSourceConfig       the http source config
-     * @param httpClient             the http client
-     * @param requestVariablesValues the request variables values
+     * @param httpSourceConfig        the http source config
+     * @param httpClient              the http client
+     * @param requestVariablesValues  the request variables values
+     * @param endpointVariablesValues the request variables values
      */
-    public HttpGetRequestHandler(HttpSourceConfig httpSourceConfig, AsyncHttpClient httpClient, Object[] requestVariablesValues, Object[] dynamicHeaderVariablesValues) {
+    public HttpGetRequestHandler(HttpSourceConfig httpSourceConfig, AsyncHttpClient httpClient, Object[] requestVariablesValues, Object[] dynamicHeaderVariablesValues, Object[] endpointVariablesValues) {
         this.httpSourceConfig = httpSourceConfig;
         this.httpClient = httpClient;
         this.requestVariablesValues = requestVariablesValues;
         this.dynamicHeaderVariablesValues = dynamicHeaderVariablesValues;
+        this.endpointVariablesValues = endpointVariablesValues;
     }
 
     @Override
     public BoundRequestBuilder create() {
         String endpointPath = String.format(httpSourceConfig.getPattern(), requestVariablesValues);
-        String endpoint = httpSourceConfig.getEndpoint();
+        String endpoint = String.format(httpSourceConfig.getEndpoint(), endpointVariablesValues);
+
         String requestEndpoint = endpoint + endpointPath;
         BoundRequestBuilder getRequest = httpClient.prepareGet(requestEndpoint);
         Map<String, String> headers = httpSourceConfig.getHeaders();

diff --git a/...ain/java/io/odpf/dagger/core/processors/external/http/request/HttpPostRequestHandler.java b/...ain/java/io/odpf/dagger/core/processors/external/http/request/HttpPostRequestHandler.java
@@ -19,24 +19,28 @@ public class HttpPostRequestHandler implements HttpRequestHandler {
     private AsyncHttpClient httpClient;
     private Object[] requestVariablesValues;
     private Object[] dynamicHeaderVariablesValues;
+    private Object[] endpointVariablesValues;
     /**
      * Instantiates a new Http post request handler.
      *
-     * @param httpSourceConfig       the http source config
-     * @param httpClient             the http client
-     * @param requestVariablesValues the request variables values
+     * @param httpSourceConfig        the http source config
+     * @param httpClient              the http client
+     * @param requestVariablesValues  the request variables values
+     * @param endpointVariablesValues the endpoint variables values
      */
-    public HttpPostRequestHandler(HttpSourceConfig httpSourceConfig, AsyncHttpClient httpClient, Object[] requestVariablesValues, Object[] dynamicHeaderVariablesValues) {
+    public HttpPostRequestHandler(HttpSourceConfig httpSourceConfig, AsyncHttpClient httpClient, Object[] requestVariablesValues, Object[] dynamicHeaderVariablesValues, Object[] endpointVariablesValues) {
         this.httpSourceConfig = httpSourceConfig;
         this.httpClient = httpClient;
         this.requestVariablesValues = requestVariablesValues;
         this.dynamicHeaderVariablesValues = dynamicHeaderVariablesValues;
+        this.endpointVariablesValues = endpointVariablesValues;
     }
 
     @Override
     public BoundRequestBuilder create() {
         String requestBody = String.format(httpSourceConfig.getPattern(), requestVariablesValues);
-        String endpoint = httpSourceConfig.getEndpoint();
+        String endpoint = String.format(httpSourceConfig.getEndpoint(), endpointVariablesValues);
+
         BoundRequestBuilder postRequest = httpClient
                 .preparePost(endpoint)
                 .setBody(requestBody);

diff --git a/...main/java/io/odpf/dagger/core/processors/external/http/request/HttpPutRequestHandler.java b/...main/java/io/odpf/dagger/core/processors/external/http/request/HttpPutRequestHandler.java
@@ -0,0 +1,65 @@
+package io.odpf.dagger.core.processors.external.http.request;
+
+import com.google.gson.Gson;
+import io.netty.util.internal.StringUtil;
+import io.odpf.dagger.core.exception.InvalidConfigurationException;
+import io.odpf.dagger.core.processors.external.http.HttpSourceConfig;
+import org.asynchttpclient.AsyncHttpClient;
+import org.asynchttpclient.BoundRequestBuilder;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.UnknownFormatConversionException;
+
+/**
+ * The Http post request handler.
+ */
+public class HttpPutRequestHandler implements HttpRequestHandler {
+    private HttpSourceConfig httpSourceConfig;
+    private AsyncHttpClient httpClient;
+    private Object[] requestVariablesValues;
+    private Object[] dynamicHeaderVariablesValues;
+    private Object[] endpointVariablesValues;
+    /**
+     * Instantiates a new Http post request handler.
+     *
+     * @param httpSourceConfig        the http source config
+     * @param httpClient              the http client
+     * @param requestVariablesValues  the request variables values
+     * @param endpointVariablesValues the endpoint variables values
+     */
+    public HttpPutRequestHandler(HttpSourceConfig httpSourceConfig, AsyncHttpClient httpClient, Object[] requestVariablesValues, Object[] dynamicHeaderVariablesValues, Object[] endpointVariablesValues) {
+        this.httpSourceConfig = httpSourceConfig;
+        this.httpClient = httpClient;
+        this.requestVariablesValues = requestVariablesValues;
+        this.dynamicHeaderVariablesValues = dynamicHeaderVariablesValues;
+        this.endpointVariablesValues = endpointVariablesValues;
+    }
+
+    @Override
+    public BoundRequestBuilder create() {
+        String requestBody = String.format(httpSourceConfig.getPattern(), requestVariablesValues);
+        String endpoint = String.format(httpSourceConfig.getEndpoint(), endpointVariablesValues);
+
+        BoundRequestBuilder putRequest = httpClient
+                .preparePut(endpoint)
+                .setBody(requestBody);
+        Map<String, String> headers = httpSourceConfig.getHeaders();
+        if (!StringUtil.isNullOrEmpty(httpSourceConfig.getHeaderPattern())) {
+            try {
+                String dynamicHeader = String.format(httpSourceConfig.getHeaderPattern(), dynamicHeaderVariablesValues);
+                headers.putAll(new Gson().fromJson(dynamicHeader, HashMap.class));
+            } catch (UnknownFormatConversionException e) {
+                throw new InvalidConfigurationException(String.format("pattern config '%s' is invalid", httpSourceConfig.getHeaderPattern()));
+            } catch (IllegalArgumentException e) {
+                throw new InvalidConfigurationException(String.format("pattern config '%s' is incompatible with the variable config '%s'", httpSourceConfig.getHeaderPattern(), httpSourceConfig.getHeaderVariables()));
+            }
+        }
+        return addHeaders(putRequest, headers);
+    }
+
+    @Override
+    public boolean canCreate() {
+        return httpSourceConfig.getVerb().equalsIgnoreCase("put");
+    }
+}
diff --git a/...rc/main/java/io/odpf/dagger/core/processors/external/http/request/HttpRequestFactory.java b/...rc/main/java/io/odpf/dagger/core/processors/external/http/request/HttpRequestFactory.java
@@ -19,11 +19,12 @@ public class HttpRequestFactory {
      * @param requestVariablesValues the request variables values
      * @return the bound request builder
      */
-    public static BoundRequestBuilder createRequest(HttpSourceConfig httpSourceConfig, AsyncHttpClient httpClient, Object[] requestVariablesValues,  Object[] headerVariablesValues) {
+    public static BoundRequestBuilder createRequest(HttpSourceConfig httpSourceConfig, AsyncHttpClient httpClient, Object[] requestVariablesValues,  Object[] headerVariablesValues, Object[] endpointVariablesValues) {
 
         ArrayList<HttpRequestHandler> httpRequestHandlers = new ArrayList<>();
-        httpRequestHandlers.add(new HttpPostRequestHandler(httpSourceConfig, httpClient, requestVariablesValues, headerVariablesValues));
-        httpRequestHandlers.add(new HttpGetRequestHandler(httpSourceConfig, httpClient, requestVariablesValues, headerVariablesValues));
+        httpRequestHandlers.add(new HttpPostRequestHandler(httpSourceConfig, httpClient, requestVariablesValues, headerVariablesValues, endpointVariablesValues));
+        httpRequestHandlers.add(new HttpGetRequestHandler(httpSourceConfig, httpClient, requestVariablesValues, headerVariablesValues, endpointVariablesValues));
+        httpRequestHandlers.add(new HttpPutRequestHandler(httpSourceConfig, httpClient, requestVariablesValues, headerVariablesValues, endpointVariablesValues));
 
         HttpRequestHandler httpRequestHandler = httpRequestHandlers
                 .stream()

diff --git a/dagger-core/src/test/java/io/odpf/dagger/core/processors/PostProcessorConfigTest.java b/dagger-core/src/test/java/io/odpf/dagger/core/processors/PostProcessorConfigTest.java
@@ -64,7 +64,7 @@ public void shouldReturnHttpExternalSourceConfig() {
         outputMapping = new OutputMapping("$.data.tensor.values[0]");
         outputMappings.put("surge_factor", outputMapping);
 
-        HttpSourceConfig httpSourceConfig = new HttpSourceConfig("http://localhost:8000", "post", null, null, null, null, "5000", "5000", true, null, null, headerMap, outputMappings, null, false);
+        HttpSourceConfig httpSourceConfig = new HttpSourceConfig("http://localhost:8000", "", "post", null, null, null, null, "5000", "5000", true, null, null, headerMap, outputMappings, null, false);
 
         assertEquals(httpSourceConfig, defaultPostProcessorConfig.getExternalSource().getHttpConfig().get(0));
     }
@@ -120,7 +120,7 @@ public void shouldBeEmptyWhenNoneOfTheConfigsExist() {
     @Test
     public void shouldNotBeEmptyWhenExternalSourceHasHttpConfigExist() {
         ArrayList<HttpSourceConfig> http = new ArrayList<>();
-        http.add(new HttpSourceConfig("", "", "", "", "", "", "", "", false, "", "", new HashMap<>(), new HashMap<>(), "metricId_01", false));
+        http.add(new HttpSourceConfig("", "", "", "", "", "", "", "", "", false, "", "", new HashMap<>(), new HashMap<>(), "metricId_01", false));
         ArrayList<EsSourceConfig> es = new ArrayList<>();
         ArrayList<PgSourceConfig> pg = new ArrayList<>();
         ExternalSourceConfig externalSourceConfig = new ExternalSourceConfig(http, es, pg, new ArrayList<>());
@@ -290,4 +290,18 @@ public void shouldParseInternalProcessorConfigForInternalSourceConfig() {
         PostProcessorConfig postProcessorConfig = PostProcessorConfig.parse(configuration);
         assertNotNull(postProcessorConfig.getInternalSource().get(0).getInternalProcessorConfig());
     }
+
+    @Test
+    public void shouldParseEndpointVariablesConfig() {
+        String configuration = "{\"external_source\":{\"es\":[{\"host\":\"localhost\",\"port\":\"9200\",\"output_mapping\":{\"customer_profile\":{\"path\":\"$._source\"}},\"endpoint_pattern\":\"/customers/customer/%s\",\"endpoint_variables\":\"customer_id\",\"retry_timeout\":\"5000\",\"socket_timeout\":\"6000\",\"stream_timeout\":\"5000\",\"type\":\"TestLogMessage\"}],\"http\":[{\"body_column_from_sql\":\"request_body\",\"connect_timeout\":\"5000\",\"endpoint\":\"http://localhost:8000/%s\",\"endpoint_variables\":\"some-id\",\"fail_on_errors\":\"true\",\"headers\":{\"content-type\":\"application/json\"},\"output_mapping\":{\"surge_factor\":{\"path\":\"$.data.tensor.values[0]\"}},\"stream_timeout\":\"5000\",\"verb\":\"put\"}]},\"internal_source\":[{\"output_field\":\"event_timestamp\",\"value\":\"CURRENT_TIMESTAMP\",\"type\":\"function\"},{\"output_field\":\"s2_id_level\",\"value\":\"7\",\"type\":\"constant\"}],\"transformers\":[{\"transformation_arguments\":{\"keyColumnName\":\"s2id\",\"valueColumnName\":\"features\"},\"transformation_class\":\"test.postprocessor.FeatureTransformer\"}]}";
+        PostProcessorConfig postProcessorConfig = PostProcessorConfig.parse(configuration);
+        assertEquals("some-id", postProcessorConfig.getExternalSource().getHttpConfig().get(0).getEndpointVariables());
+    }
+
+    @Test
+    public void shouldParseEmptyEndpointVariablesConfig() {
+        String configuration = "{\"external_source\":{\"es\":[{\"host\":\"localhost\",\"port\":\"9200\",\"output_mapping\":{\"customer_profile\":{\"path\":\"$._source\"}},\"endpoint_pattern\":\"/customers/customer/%s\",\"endpoint_variables\":\"customer_id\",\"retry_timeout\":\"5000\",\"socket_timeout\":\"6000\",\"stream_timeout\":\"5000\",\"type\":\"TestLogMessage\"}],\"http\":[{\"body_column_from_sql\":\"request_body\",\"connect_timeout\":\"5000\",\"endpoint\":\"http://localhost:8000/%s\",\"fail_on_errors\":\"true\",\"headers\":{\"content-type\":\"application/json\"},\"output_mapping\":{\"surge_factor\":{\"path\":\"$.data.tensor.values[0]\"}},\"stream_timeout\":\"5000\",\"verb\":\"put\"}]},\"internal_source\":[{\"output_field\":\"event_timestamp\",\"value\":\"CURRENT_TIMESTAMP\",\"type\":\"function\"},{\"output_field\":\"s2_id_level\",\"value\":\"7\",\"type\":\"constant\"}],\"transformers\":[{\"transformation_arguments\":{\"keyColumnName\":\"s2id\",\"valueColumnName\":\"features\"},\"transformation_class\":\"test.postprocessor.FeatureTransformer\"}]}";
+        PostProcessorConfig postProcessorConfig = PostProcessorConfig.parse(configuration);
+        assertEquals(null, postProcessorConfig.getExternalSource().getHttpConfig().get(0).getEndpointVariables());
+    }
 }
diff --git a/...core/src/test/java/io/odpf/dagger/core/processors/external/ExternalPostProcessorTest.java b/...core/src/test/java/io/odpf/dagger/core/processors/external/ExternalPostProcessorTest.java
@@ -72,7 +72,7 @@ public void setup() {
         HashMap<String, OutputMapping> httpColumnNames = new HashMap<>();
         httpColumnNames.put("http_field_1", new OutputMapping(""));
         httpColumnNames.put("http_field_2", new OutputMapping(""));
-        HttpSourceConfig httpSourceConfig = new HttpSourceConfig("endpoint", "POST", "/some/patttern/%s", "variable", "", "", "123", "234", false, "type", "20", new HashMap<>(), httpColumnNames, "metricId_01", false);
+        HttpSourceConfig httpSourceConfig = new HttpSourceConfig("endpoint", "", "POST", "/some/patttern/%s", "variable", "", "", "123", "234", false, "type", "20", new HashMap<>(), httpColumnNames, "metricId_01", false);
         HashMap<String, OutputMapping> esOutputMapping = new HashMap<>();
         esOutputMapping.put("es_field_1", new OutputMapping(""));
         EsSourceConfig esSourceConfig = new EsSourceConfig("host", "port", "", "", "endpointPattern",
@@ -135,7 +135,7 @@ public void shouldProcessWithRightConfiguration() {
         outputMapping.put("order_id", new OutputMapping("path"));
 
         List<HttpSourceConfig> httpSourceConfigs = new ArrayList<>();
-        HttpSourceConfig httpSourceConfig = new HttpSourceConfig("endpoint", "POST", "/some/patttern/%s", "variable", "", "", "123", "234", false, "type", "20", new HashMap<>(), outputMapping, "metricId_01", false);
+        HttpSourceConfig httpSourceConfig = new HttpSourceConfig("endpoint", "", "POST", "/some/patttern/%s", "variable", "", "", "123", "234", false, "type", "20", new HashMap<>(), outputMapping, "metricId_01", false);
         httpSourceConfigs.add(httpSourceConfig);
 
         List<EsSourceConfig> esSourceConfigs = new ArrayList<>();

diff --git a/...-core/src/test/java/io/odpf/dagger/core/processors/external/ExternalSourceConfigTest.java b/...-core/src/test/java/io/odpf/dagger/core/processors/external/ExternalSourceConfigTest.java
@@ -29,7 +29,7 @@ public void setUp() {
         HashMap<String, OutputMapping> httpOutputMapping = new HashMap<>();
         httpOutputMapping.put("http_field_1", new OutputMapping(""));
         httpOutputMapping.put("http_field_2", new OutputMapping(""));
-        HttpSourceConfig httpSourceConfig = new HttpSourceConfig("endpoint", "POST", "/some/patttern/%s", "variable", "", "", "123", "234", false, "type", "20", new HashMap<>(), httpOutputMapping, "metricId_01", false);
+        HttpSourceConfig httpSourceConfig = new HttpSourceConfig("endpoint", "", "POST", "/some/patttern/%s", "variable", "", "", "123", "234", false, "type", "20", new HashMap<>(), httpOutputMapping, "metricId_01", false);
         http = new ArrayList<>();
         http.add(httpSourceConfig);
         es = new ArrayList<>();