Add HTTP Action plugin

docs/HTTP-action.md

@@ -0,0 +1,119 @@
+# HTTP Action
+
+Description
+-----------
+This plugin executing HTTP/HTTPS requests
+
+Properties
+----------
+
+### General
+
+**URL:** Url to fetch to the first page.
+The url must start with a protocol (e.g. http://).
+
+**HTTP Method:** HTTP request method.
+
+**Headers:** Headers to send with each HTTP request.
+
+**Request body:** Body to send with each HTTP request.
+
+
+### Basic Authentication
+
+**Username:** Username for basic authentication.
+
+**Password:** Password for basic authentication.
+
+### HTTP Proxy
+
+**Proxy URL:** Proxy URL. Must contain a protocol, address and port.
+
+**Username:** Proxy username.
+
+**Password:** Proxy password.
+
+### Error Handling
+
+**HTTP Errors Handling:** Defines the error handling strategy to use for certain HTTP response codes.
+The left column contains a regular expression for HTTP status code. The right column contains an action which
+is done in case of match. If HTTP status code matches multiple regular expressions, the first specified in mapping
+is matched.
+
+Example:
+
+| HTTP Code Regexp  | Error Handling          |
+| ----------------- |:-----------------------:|
+| 2..               | Success                 |
+| 401               | Retry and fail          |
+| 4..               | Fail                    |
+| 5..               | Retry and send to error |
+| .*                | Fail                    |
+
+Note: pagination types "Link in response header", "Link in response body", "Token in response body" do not support
+"Send to error", "Skip", "Retry and send to error", "Retry and skip" options.
+
+**Non-HTTP Error Handling:** Error handling strategy to use when the HTTP response cannot be transformed to an output record.
+Possible values are:<br>
+Stop on error - Fails pipeline due to erroneous record.<br>
+Send to error - Sends erroneous record's text to error port and continues.<br>
+Skip on error - Ignores erroneous records.
+
+**Connect Timeout:** Maximum time in seconds connection initialization is allowed to take.
+
+### OAuth2
+
+**OAuth2 Enabled:** If true, plugin will perform OAuth2 authentication.
+
+**Auth URL:** Endpoint for the authorization server used to retrieve the authorization code.
+
+**Token URL:** Endpoint for the resource server, which exchanges the authorization code for an access token.
+
+**Client ID:** Client identifier obtained during the Application registration process.
+
+**Client Secret:** Client secret obtained during the Application registration process.
+
+**Scopes:** Scope of the access request, which might have multiple space-separated values.
+
+**Refresh Token:** Token used to receive accessToken, which is end product of OAuth2.
+
+### SSL/TLS
+
+**Verify HTTPS Trust Certificates:** If false, untrusted trust certificates (e.g. self signed), will not lead to an
+error. Do not disable this in production environment on a network you do not entirely trust. Especially public internet.
+
+**Keystore File:** A path to a file which contains keystore.
+
+**Keystore Type:** Format of a keystore.
+
+**Keystore Password:** Password for a keystore. If a keystore is not password protected leave it empty.
+
+**Keystore Key Algorithm:** An algorithm used for keystore.
+
+**Keystore Cert Alias**
+
+ Alias of the key in the keystore to be used for communication. This options is supported only by X.509 keys or keystores.
+
+Below is an example how the store need to be prepared:
+ ```
+ cat  client.crt client.key > client-bundle.pem
+
+ openssl pkcs12 -export -in client-bundle.pem -out full-chain.keycert.p12 -name ${CERT_ALIAS}
+
+ keytool -importkeystore -srckeystore full-chain.keycert.p12 -srcstoretype pkcs12 -srcalias ${CERT_ALIAS} \
+                         -destkeystore identity.jks -deststoretype jks -destalias ${CERT_ALIAS}
+ ```
+
+**TrustStore File:** A path to a file which contains truststore.
+
+**TrustStore Type:** Format of a truststore.
+
+**TrustStore Password:** Password for a truststore. If a truststore is not password protected leave it empty.
+
+**TrustStore Key Algorithm:** An algorithm used for truststore.
+
+**Transport Protocols:** Transport protocols which are allowed for connection.
+
+**Cipher Suites:** Cipher suites which are allowed for connection.
+Colons, commas or spaces are also acceptable separators.
+

docs/HTTP-batchsource.md

@@ -403,6 +403,20 @@ error. Do not disable this in production environment on a network you do not ent
 
 **Keystore Key Algorithm:** An algorithm used for keystore.
 
+**Keystore Cert Alias**
+
+ Alias of the key in the keystore to be used for communication. This options is supported only by X.509 keys or keystores.
+
+Below is an example how the store need to be prepared:
+ ```
+ cat  client.crt client.key > client-bundle.pem
+
+ openssl pkcs12 -export -in client-bundle.pem -out full-chain.keycert.p12 -name ${CERT_ALIAS}
+
+ keytool -importkeystore -srckeystore full-chain.keycert.p12 -srcstoretype pkcs12 -srcalias ${CERT_ALIAS} \
+                         -destkeystore identity.jks -deststoretype jks -destalias ${CERT_ALIAS}
+ ```
+
 **TrustStore File:** A path to a file which contains truststore.
 
 **TrustStore Type:** Format of a truststore.

docs/HTTP-streamingsource.md

@@ -410,6 +410,20 @@ error. Do not disable this in production environment on a network you do not ent
 
 **Keystore Key Algorithm:** An algorithm used for keystore.
 
+**Keystore Cert Alias**
+
+Alias of the key in the keystore to be used for communication. This options is supported only by X.509 keys or keystores.
+
+Below is an example how the store need to be prepared:
+ ```
+ cat  client.crt client.key > client-bundle.pem
+
+ openssl pkcs12 -export -in client-bundle.pem -out full-chain.keycert.p12 -name ${CERT_ALIAS}
+
+ keytool -importkeystore -srckeystore full-chain.keycert.p12 -srcstoretype pkcs12 -srcalias ${CERT_ALIAS} \
+                         -destkeystore identity.jks -deststoretype jks -destalias ${CERT_ALIAS}
+ ```
+
 **TrustStore File:** A path to a file which contains truststore.
 
 **TrustStore Type:** Format of a truststore.

pom.xml

@@ -21,7 +21,7 @@
   <name>HTTP Plugins</name>
   <groupId>io.cdap</groupId>
   <artifactId>http-plugins</artifactId>
-  <version>1.4.0-SNAPSHOT</version>
+  <version>1.4.2-SNAPSHOT</version>
 
   <licenses>
     <license>
@@ -79,10 +79,12 @@
     <cdap.version>6.1.1</cdap.version>
     <commons.version>3.9</commons.version>
     <common.codec.version>1.12</common.codec.version>
+    <common.logging.version>1.2</common.logging.version>
+    <log4j.version>1.2.17</log4j.version>
     <gson.version>2.8.5</gson.version>
     <hadoop.version>2.3.0</hadoop.version>
     <httpcomponents.version>4.5.9</httpcomponents.version>
-    <hydrator.version>2.4.0-SNAPSHOT</hydrator.version>
+    <hydrator.version>2.4.0</hydrator.version>
     <jackson.version>2.9.9</jackson.version>
     <junit.version>4.11</junit.version>
     <jython.version>2.7.1</jython.version>
@@ -93,6 +95,20 @@
   </properties>
 
   <dependencies>
+    <!-- Required to work properly on external hadoop clusters-->
+    <dependency>
+      <groupId>commons-logging</groupId>
+      <artifactId>commons-logging</artifactId>
+      <version>${common.logging.version}</version>
+      <scope>compile</scope>
+    </dependency>
+    <dependency>
+     <groupId>log4j</groupId>
+     <artifactId>log4j</artifactId>
+     <version>${log4j.version}</version>
+     <scope>compile</scope>
+    </dependency>
+    <!-- / end required -->
     <dependency>
       <groupId>io.cdap.cdap</groupId>
       <artifactId>cdap-api</artifactId>

src/main/java/io/cdap/plugin/http/action/HttpAction.java

@@ -0,0 +1,101 @@
+/*
+ * Copyright © 2021 Cask Data, Inc.
+ *
+ * Licensed 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 io.cdap.plugin.http.action;
+
+import io.cdap.cdap.api.annotation.Description;
+import io.cdap.cdap.api.annotation.Name;
+import io.cdap.cdap.api.annotation.Plugin;
+import io.cdap.cdap.etl.api.PipelineConfigurer;
+import io.cdap.cdap.etl.api.action.Action;
+import io.cdap.cdap.etl.api.action.ActionContext;
+import io.cdap.plugin.http.common.RetryPolicy;
+import io.cdap.plugin.http.common.error.HttpErrorHandler;
+import io.cdap.plugin.http.common.error.RetryableErrorHandling;
+import io.cdap.plugin.http.common.http.HttpClient;
+import io.cdap.plugin.http.common.http.HttpConstants;
+import io.cdap.plugin.http.common.http.HttpResponse;
+import org.awaitility.Awaitility;
+import org.awaitility.core.ConditionTimeoutException;
+import org.awaitility.pollinterval.FixedPollInterval;
+import org.awaitility.pollinterval.IterativePollInterval;
+import org.awaitility.pollinterval.PollInterval;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * HTTP Action Plugin
+ */
+@Plugin(type = Action.PLUGIN_TYPE)
+@Name(HttpConstants.HTTP_PLUGIN_NAME)
+@Description("Action that runs a HTTP command")
+public class HttpAction extends Action {
+
+  private static final Logger LOG = LoggerFactory.getLogger(HttpAction.class);
+  private final HttpActionConfig config;
+
+  public HttpAction(HttpActionConfig config) {
+    this.config = config;
+  }
+
+    private boolean makeRequest(HttpClient httpClient, String uri, HttpErrorHandler errorHandler) throws IOException {
+      try (HttpResponse response = new HttpResponse(httpClient.executeHTTP(uri))) {
+        RetryableErrorHandling errorHandlingStrategy = errorHandler.getErrorHandlingStrategy(response.getStatusCode());
+
+        if (errorHandlingStrategy == RetryableErrorHandling.FAIL) {
+          String body = "null";
+          try {
+            body = response.getBody();
+          } catch (Exception ignore) {
+          }
+          LOG.warn(
+            String.format("Request to the url '%s' failed with error code %s and message: %s",
+                    uri,
+                    response.getStatusCode(),
+                    body
+          ));
+        }
+        return !errorHandlingStrategy.shouldRetry();
+      }
+  }
+
+  @Override
+  public void run(ActionContext actionContext) throws Exception {
+    try (HttpClient httpClient = new HttpClient(config)) {
+      HttpErrorHandler httpErrorHandler = new HttpErrorHandler(config);
+      PollInterval pollInterval = (config.getRetryPolicy().equals(RetryPolicy.LINEAR))
+        ? FixedPollInterval.fixed(Objects.requireNonNull(config.getLinearRetryInterval()), TimeUnit.SECONDS)
+        : IterativePollInterval.iterative(duration -> duration.multiply(2));
+
+      Awaitility
+        .await().with()
+        .pollInterval(pollInterval)
+        .timeout(config.getMaxRetryDuration(), TimeUnit.SECONDS)
+        .until(() -> makeRequest(httpClient, config.getUrl(), httpErrorHandler));
+    } catch (ConditionTimeoutException e) {
+      LOG.warn(String.format("Request to the url '%s' failed due to timeout", config.getUrl()));
+    }
+  }
+
+  @Override
+  public void configurePipeline(PipelineConfigurer pipelineConfigurer) {
+    super.configurePipeline(pipelineConfigurer);
+    config.validate();
+  }
+}