Merge pull request #80 from data-integrations/PLUGIN-692

[PLUGIN-692] Add Service Account auth to http

Author: sumengwang

docs/HTTP-batchsource.md

@@ -203,11 +203,30 @@ can be omitted as long as the field is present in schema.
 
 **CSV Skip First Row:** Whether to skip the first row of the HTTP response. This is usually set if the first row is a header row.
 
-### Basic Authentication
-
-**Username:** Username for basic authentication.
-
-**Password:** Password for basic authentication.
+### Authentication
+* **OAuth2**
+    * **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.
+* **Service Account** - service account key used for authorization
+    * **File Path**: Path on the local file system of the service account key used for
+      authorization. Can be set to 'auto-detect' when running on a Dataproc cluster.
+      When running on other clusters, the file must be present on every node in the cluster.
+    * **JSON**: Contents of the service account JSON file.
+    * **Scope**: The additional Google credential scopes required to access entered url, cloud-platform is included by 
+      default, visit https://developers.google.com/identity/protocols/oauth2/scopes for more information.
+      * Scope example:
+      
+```
+https://www.googleapis.com/auth/bigquery
+https://www.googleapis.com/auth/cloud-platform
+```
+* **Basic Authentication**
+  * **Username:** Username for basic authentication.
+  * **Password:** Password for basic authentication.
 
 ### HTTP Proxy
 
@@ -373,22 +392,6 @@ def get_next_page_url(url, page, headers):
 The above code iterates over first five pages of searchcode.com results. When 'None' is returned the iteration
 is stopped.
 
-### 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

docs/HTTP-streamingsource.md

@@ -207,11 +207,30 @@ can be omitted as long as the field is present in schema.
 
 **CSV Skip First Row:** Whether to skip the first row of the HTTP response. This is usually set if the first row is a header row.
 
-### Basic Authentication
-
-**Username:** Username for basic authentication.
-
-**Password:** Password for basic authentication.
+### Authentication
+* **OAuth2**
+    * **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.
+* **Service Account** - service account key used for authorization
+    * **File Path**: Path on the local file system of the service account key used for
+      authorization. Can be set to 'auto-detect' when running on a Dataproc cluster.
+      When running on other clusters, the file must be present on every node in the cluster.
+    * **JSON**: Contents of the service account JSON file.
+    * **Scope**: The additional Google credential scopes required to access entered url, cloud-platform is included by
+      default, visit https://developers.google.com/identity/protocols/oauth2/scopes for more information.
+        * Scope example:
+      
+```
+https://www.googleapis.com/auth/bigquery
+https://www.googleapis.com/auth/cloud-platform
+```
+* **Basic Authentication**
+    * **Username:** Username for basic authentication.
+    * **Password:** Password for basic authentication.
 
 ### HTTP Proxy
 
@@ -381,22 +400,6 @@ def get_next_page_url(url, page, headers):
 The above code iterates over first five pages of searchcode.com results. When 'None' is returned the iteration
 is stopped.
 
-### 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

pom.xml

@@ -76,13 +76,14 @@
 
   <properties>
     <awaitility.version>3.1.6</awaitility.version>
-    <cdap.version>6.1.1</cdap.version>
+    <cdap.version>6.8.0-SNAPSHOT</cdap.version>
     <commons.version>3.9</commons.version>
     <common.codec.version>1.12</common.codec.version>
     <gson.version>2.8.5</gson.version>
+    <googleauth.version>0.1.0</googleauth.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.10.0-SNAPSHOT</hydrator.version>
     <jackson.version>2.9.9</jackson.version>
     <junit.version>4.11</junit.version>
     <jython.version>2.7.1</jython.version>
@@ -324,6 +325,11 @@
       <artifactId>httpclient</artifactId>
       <version>${httpcomponents.version}</version>
     </dependency>
+    <dependency>
+      <groupId>com.google.auth</groupId>
+      <artifactId>google-auth-library-oauth2-http</artifactId>
+      <version>${googleauth.version}</version>
+    </dependency>
     <dependency>
       <groupId>com.google.code.gson</groupId>
       <artifactId>gson</artifactId>

src/main/java/io/cdap/plugin/http/source/batch/HttpBatchSource.java

@@ -25,6 +25,7 @@
 import io.cdap.cdap.api.data.schema.Schema;
 import io.cdap.cdap.api.dataset.lib.KeyValue;
 import io.cdap.cdap.etl.api.Emitter;
+import io.cdap.cdap.etl.api.FailureCollector;
 import io.cdap.cdap.etl.api.InvalidEntry;
 import io.cdap.cdap.etl.api.PipelineConfigurer;
 import io.cdap.cdap.etl.api.batch.BatchRuntimeContext;
@@ -59,15 +60,17 @@ public HttpBatchSource(HttpBatchSourceConfig config) {
 
   @Override
   public void configurePipeline(PipelineConfigurer pipelineConfigurer) {
-    config.validate(); // validate when macros not yet substituted
+    FailureCollector failureCollector = pipelineConfigurer.getStageConfigurer().getFailureCollector();
+    config.validate(failureCollector); // validate when macros not yet substituted
     config.validateSchema();
 
     pipelineConfigurer.getStageConfigurer().setOutputSchema(config.getSchema());
   }
 
   @Override
   public void prepareRun(BatchSourceContext context) {
-    config.validate(); // validate when macros are already substituted
+    FailureCollector failureCollector = context.getFailureCollector();
+    config.validate(failureCollector); // validate when macros are already substituted
     config.validateSchema();
 
     schema = config.getSchema();

src/main/java/io/cdap/plugin/http/source/batch/HttpBatchSourceConfig.java

@@ -39,6 +39,7 @@ private HttpBatchSourceConfig(HttpBatchSourceConfigBuilder builder) {
     this.readTimeout = builder.readTimeout;
     this.paginationType = builder.paginationType;
     this.verifyHttps = builder.verifyHttps;
+    this.authType = builder.authType;
   }
 
   public static HttpBatchSourceConfigBuilder builder() {
@@ -63,6 +64,7 @@ public static class HttpBatchSourceConfigBuilder {
     private Integer readTimeout;
     private String paginationType;
     private String verifyHttps;
+    private String authType;
 
     public HttpBatchSourceConfigBuilder setReferenceName (String referenceName) {
       this.referenceName = referenceName;
@@ -93,7 +95,7 @@ public HttpBatchSourceConfigBuilder setOauth2Enabled(String oauth2Enabled) {
       this.oauth2Enabled = oauth2Enabled;
       return this;
     }
-
+    
     public HttpBatchSourceConfigBuilder setErrorHandling(String errorHandling) {
       this.errorHandling = errorHandling;
       return this;
@@ -129,6 +131,11 @@ public HttpBatchSourceConfigBuilder setVerifyHttps(String verifyHttps) {
       return this;
     }
 
+    public HttpBatchSourceConfigBuilder setAuthType(String authType) {
+      this.authType = authType;
+      return this;
+    }
+
     public HttpBatchSourceConfig build() {
       return new HttpBatchSourceConfig(this);
     }