Merge pull request #43 from GoogleCloudPlatform/gcs-s3-inspect

gcs-s3-dlp-inspect cloud build

Author: santhh

README.md

@@ -535,9 +535,9 @@ DLP Inspect template:
 
 ```
 
-### DLP Inspection for Data Stored in AWS S3 Bucket
+### DLP Inspection for Data Stored in AWS S3 and GCS Bucket
 
-This PoC can be used to inspect large scale dataset (csv, txt) stored in AWS S3 bucket. It uses Dataflow S3 connector, invoke DLP Inspect API to inspect data based on some configuration specified in a DLP inspect template. It stores the result in BQ.   
+This PoC can be used to inspect large scale dataset (csv, txt) stored in AWS S3 and GCS buckets. It uses Dataflow S3 connector, invoke DLP Inspect API to inspect data based on some configuration specified in a DLP inspect template. It stores the result in BQ.   
 ### How it works?
 1. Build and Run the pipline in a GCP project using Dataflow. Please ensure you have enabled DLP and DF apis before executing.
 
@@ -561,9 +561,79 @@ gradle build -DmainClass=com.google.swarm.tokenization.S3Import --x test
 
 To Run: 
 
-gradle run -DmainClass=com.google.swarm.tokenization.S3Import -Pargs=" --streaming --project=<id> --runner=DataflowRunner --awsAccessKey=<key>--awsSecretKey=<key>--s3BucketUrl=s3://<bucket>/*.* --inspectTemplateName=projects/<id>/inspectTemplates/<template_id> --awsRegion=ca-central-1 --numWorkers=50 --workerMachineType=n1-highmem-16 --maxNumWorkers=50 --autoscalingAlgorithm=NONE --enableStreamingEngine --tempLocation=gs://<bucket>/temp --dataSetId=<dataset_id> --s3ThreadPoolSize=1000 --maxConnections=1000000 --socketTimeout=100 --connectionTimeout=100"
+gradle run -DmainClass=com.google.swarm.tokenization.S3Import -Pargs=" --streaming --project=<id> --runner=DataflowRunner --awsAccessKey=<key>--awsSecretKey=<key>--s3BucketUrl=s3://<bucket>/*.* --gcsBucketUrl=gs://<bucket>/*.* --inspectTemplateName=projects/<id>/inspectTemplates/<template_id> --awsRegion=ca-central-1 --numWorkers=50 --workerMachineType=n1-highmem-16 --maxNumWorkers=50 --autoscalingAlgorithm=NONE --enableStreamingEngine --tempLocation=gs://<bucket>/temp --dataSetId=<dataset_id> --s3ThreadPoolSize=1000 --maxConnections=1000000 --socketTimeout=100 --connectionTimeout=100"
 
 ```
+
+### Run Using Cloud Build 
+1. Modify inspect config file (gcs-s3-inspect-config.json) to add/update the info types you would lilke to use for scan. Below snippet is a sample config file used for demo.
+
+```
+{
+	"inspectTemplate": {
+		"displayName": "DLP Inspection Config For Demo",
+		"description": "DLP Config To Inspect GCS and S3 Bucket",
+		"inspectConfig": {
+			"infoTypes": [
+				{
+					"name": "EMAIL_ADDRESS"
+				},
+				{
+					"name": "CREDIT_CARD_NUMBER"
+				},
+				{
+					"name": "PHONE_NUMBER"
+				},
+				{
+					"name": "US_SOCIAL_SECURITY_NUMBER"
+				},
+				{
+					"name": "IP_ADDRESS"
+				}
+			],
+			"minLikelihood": "POSSIBLE",
+			"customInfoTypes": [
+				{
+					"infoType": {
+						"name": "ONLINE_USER_ID"
+					},
+					"regex": {
+						"pattern": "\\b:\\d{16}"
+					}
+				}
+			]
+		}
+	},
+}
+
+```
+
+2. Export Required Parameters
+
+```
+export AWS_ACCESS_KEY=<aws_access_key>
+export AWS_SECRET_KEY=<aws_secret_key>
+export S3_BUCKET_URL=<s3_bucket_url>
+export GCS_BUCKET_URL=<gcs_bucket_url>
+export AWS_REGION=<aws_region>
+export BQ_DATASET=<bq_dataset>
+    
+```
+
+3. Run Cloud Build command
+
+``` 
+gcloud builds submit . --config dlp-demo-s3-gcs-inspect.yaml \ 
+--substitutions _AWS_ACCESS_KEY=$AWS_ACCESS_KEY, \
+_API_KEY=$(gcloud auth print-access-token), \
+_AWS_SECRET_KEY=$AWS_SECRET_KEY, \ 
+_S3_BUCKET_URL=$S3_BUCKET_URL, \
+_GCS_BUCKET_URL=$GCS_BUCKET_URL, \ 
+_AWS_REGION=$AWS_REGION, \ 
+_BQ_DATASET=$BQ_DATASET
+
+```
+
 ### Testing Configuration
 This PoC was built to process large scale data by scaling number of workers horizontally. During our test run, we have successfully inspected 1.3 TB of data in less than 10 minutes. It's recommended to use n1-highmem-16 machines as it allows Dataflow to reserve more JVM heap memory.
 
@@ -580,6 +650,9 @@ Below configurations are common for both setup:
 
 
 ### Some Screenshots from the PoC run
+#### Dataflow Job
+![Dataflow_DAG](diagrams/s3_1.png)
+![Dataflow_DAG](diagrams/s3_2.png)
 
 #### S3 Bucket
 

build.gradle

@@ -45,7 +45,17 @@ apply plugin: "com.diffplug.gradle.spotless"
 
 mainClassName = System.getProperty("mainClass")
 //mainClassName = 'com.google.swarm.tokenization.tool.DLPTemplateHelper' 
-
+jib {
+  from {
+    image = 'gcr.io/dataflow-templates-base/java8-template-launcher-base:latest'
+  }
+  container {
+    appRoot = '/template/df-dlp-inspect-template'
+    entrypoint = 'INHERIT'
+	environment = [DATAFLOW_JAVA_COMMAND_SPEC:'/template/df-dlp-inspect-template/resources/java_command_spec.json']
+	
+  }
+}
 group 'com.google.swarm'
 version '0.1.0'
 

create-df-template.sh

@@ -0,0 +1,87 @@
+#!/usr/bin/env bash
+# Copyright 2019 Google 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.#!/usr/bin/env bash
+# Copyright 2019 Google 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.
+ 
+set -x 
+
+PROJECT_ID=$1
+AWS_ACCESS_KEY=$2
+API_KEY=$3
+AWS_SECRET_KEY=$4
+S3_BUCKET_URL=$5
+GCS_BUCKET_URL=$6
+AWS_REGION=$7
+BQ_DATASET=$8
+INSPECT_CONFIG="@gcs-s3-inspect-config.json"
+INSPECT_TEMPLATE_OUTPUT="inspect-template.json"
+DLP_API_ROOT_URL="https://dlp.googleapis.com"
+INSPECT_TEMPLATE_API="${DLP_API_ROOT_URL}/v2/projects/${PROJECT_ID}/inspectTemplates"
+curl -X POST -H "Content-Type: application/json" \
+ -H "Authorization: Bearer ${API_KEY}" \
+ "${INSPECT_TEMPLATE_API}"`` \
+ -d "${INSPECT_CONFIG}"\
+ -o "${INSPECT_TEMPLATE_OUTPUT}"
+more ${INSPECT_TEMPLATE_OUTPUT}
+INSPECT_TEMPLATE_NAME=$(jq -c '.name' ${INSPECT_TEMPLATE_OUTPUT})
+# publicly hosted image
+DYNAMIC_TEMPLATE_BUCKET_SPEC=gs://dynamic-template/dynamic_template_dlp_inspect.json
+JOB_NAME="dlp-inspect-pipeline-`date +%Y%m%d-%H%M%S-%N`"
+echo $JOB_NAME
+GCS_STAGING_LOCATION=gs://dynamic-template/log
+TEMP_LOCATION=gs://dynamic-template/temp
+PARAMETERS_CONFIG='{  
+   "jobName":"'$JOB_NAME'",
+   "parameters":{  
+      "streaming":"true",
+	  "enableStreamingEngine":"true",
+	  "autoscalingAlgorithm":"NONE",
+      "workerMachineType": "n1-standard-8",
+      "numWorkers":"50",
+      "maxNumWorkers":"50",
+      "awsAccessKey":"'$AWS_ACCESS_KEY'",
+	  "awsSecretKey":"'$AWS_SECRET_KEY'",
+	  "s3BucketUrl":"'$S3_BUCKET_URL'",
+	  "gcsBucketUrl":"'$GCS_BUCKET_URL'",
+	  "inspectTemplateName":'$INSPECT_TEMPLATE_NAME',
+	  "s3ThreadPoolSize":"1000",
+	  "maxConnections":"1000000",
+	  "socketTimeout":"100",
+	  "connectionTimeout":"100",
+	  "tempLocation":"'$TEMP_LOCATION'",
+	  "awsRegion":"'$AWS_REGION'",
+	  "dataSetId":"'$BQ_DATASET'",	  
+	}
+}'
+DF_API_ROOT_URL="https://dataflow.googleapis.com"
+TEMPLATES_LAUNCH_API="${DF_API_ROOT_URL}/v1b3/projects/${PROJECT_ID}/templates:launch"
+curl -X POST -H "Content-Type: application/json" \
+ -H "Authorization: Bearer ${API_KEY}" \
+ "${TEMPLATES_LAUNCH_API}"`
+ `"?validateOnly=false"`
+ `"&dynamicTemplate.gcsPath=${DYNAMIC_TEMPLATE_BUCKET_SPEC}"` \
+ `"&dynamicTemplate.stagingLocation=${GCS_STAGING_LOCATION}" \
+ -d "${PARAMETERS_CONFIG}"

diagrams/s3_1.png

@@ -0,0 +1,17 @@
+
+ # Copyright 2019 Google 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. 
+steps:
+- name: 'ubuntu'
+  args: ['bash', '-c','apt-get -q update && apt-get install -qqy curl && apt-get install -qqy jq;sh create-df-template.sh ${PROJECT_ID} ${_AWS_ACCESS_KEY} ${_API_KEY} ${_AWS_SECRET_KEY} ${_S3_BUCKET_URL} ${_GCS_BUCKET_URL} ${_AWS_REGION} ${_BQ_DATASET}']

diagrams/s3_2.png

@@ -0,0 +1,36 @@
+{
+	"inspectTemplate": {
+		"displayName": "DLP Inspection Config For Demo",
+		"description": "DLP Config To Inspect GCS and S3 Bucket",
+		"inspectConfig": {
+			"infoTypes": [
+				{
+					"name": "EMAIL_ADDRESS"
+				},
+				{
+					"name": "CREDIT_CARD_NUMBER"
+				},
+				{
+					"name": "PHONE_NUMBER"
+				},
+				{
+					"name": "US_SOCIAL_SECURITY_NUMBER"
+				},
+				{
+					"name": "IP_ADDRESS"
+				}
+			],
+			"minLikelihood": "POSSIBLE",
+			"customInfoTypes": [
+				{
+					"infoType": {
+						"name": "ONLINE_USER_ID"
+					},
+					"regex": {
+						"pattern": "\\b:\\d{16}"
+					}
+				}
+			]
+		}
+	},
+}

dlp-demo-s3-gcs-inspect.yaml

@@ -51,7 +51,6 @@
 import org.apache.beam.sdk.metrics.Counter;
 import org.apache.beam.sdk.metrics.Metrics;
 import org.apache.beam.sdk.options.PipelineOptionsFactory;
-import org.apache.beam.sdk.options.ValueProvider;
 import org.apache.beam.sdk.transforms.DoFn;
 import org.apache.beam.sdk.transforms.Flatten;
 import org.apache.beam.sdk.transforms.ParDo;
@@ -116,13 +115,28 @@ public static void main(String[] args) {
                 WithKeys.of(file -> file.getMetadata().resourceId().getFilename().toString()))
             .setCoder(KvCoder.of(StringUtf8Coder.of(), ReadableFileCoder.of()));
 
+    // gcs files
+    PCollection<KV<String, ReadableFile>> gcsFiles =
+        p.apply(
+                "Poll GCS Files",
+                FileIO.match()
+                    .filepattern(options.getGcsBucketUrl())
+                    .continuously(DEFAULT_POLL_INTERVAL, Watch.Growth.never()))
+            .apply("GCS File Match", FileIO.readMatches().withCompression(Compression.AUTO))
+            .apply(
+                "Add GCS File Name as Key",
+                WithKeys.of(file -> file.getMetadata().resourceId().getFilename().toString()))
+            .setCoder(KvCoder.of(StringUtf8Coder.of(), ReadableFileCoder.of()));
+
     PCollection<KV<String, ReadableFile>> files =
-        s3Files.apply(
-            "Fixed Window",
-            Window.<KV<String, ReadableFile>>into(FixedWindows.of(WINDOW_INTERVAL))
-                .triggering(AfterWatermark.pastEndOfWindow())
-                .discardingFiredPanes()
-                .withAllowedLateness(Duration.ZERO));
+        PCollectionList.of(ImmutableList.of(gcsFiles, s3Files))
+            .apply("File List", Flatten.pCollections())
+            .apply(
+                "Fixed Window",
+                Window.<KV<String, ReadableFile>>into(FixedWindows.of(WINDOW_INTERVAL))
+                    .triggering(AfterWatermark.pastEndOfWindow())
+                    .discardingFiredPanes()
+                    .withAllowedLateness(Duration.ZERO));
 
     PCollectionTuple contents =
         files.apply(
@@ -245,12 +259,12 @@ public OffsetRangeTracker newTracker(OffsetRange range) {
   @SuppressWarnings("serial")
   public static class TokenizeData extends DoFn<KV<String, String>, KV<String, TableRow>> {
     private String projectId;
-    private ValueProvider<String> inspectTemplateName;
+    private String inspectTemplateName;
     private Builder requestBuilder;
     private final Counter numberOfBytesInspected =
         Metrics.counter(TokenizeData.class, "NumberOfBytesInspected");
 
-    public TokenizeData(String projectId, ValueProvider<String> inspectTemplateName) {
+    public TokenizeData(String projectId, String inspectTemplateName) {
       this.projectId = projectId;
       this.inspectTemplateName = inspectTemplateName;
     }
@@ -260,7 +274,7 @@ public void setup() {
       this.requestBuilder =
           InspectContentRequest.newBuilder()
               .setParent(ProjectName.of(this.projectId).toString())
-              .setInspectTemplateName(this.inspectTemplateName.get());
+              .setInspectTemplateName(this.inspectTemplateName);
     }
 
     @ProcessElement
@@ -338,18 +352,18 @@ public static class BQDestination
       extends DynamicDestinations<KV<String, TableRow>, KV<String, TableRow>> {
 
     private static final long serialVersionUID = 1L;
-    private ValueProvider<String> datasetName;
+    private String datasetName;
     private String projectId;
 
-    public BQDestination(ValueProvider<String> datasetName, String projectId) {
+    public BQDestination(String datasetName, String projectId) {
       this.datasetName = datasetName;
       this.projectId = projectId;
     }
 
     @Override
     public KV<String, TableRow> getDestination(ValueInSingleWindow<KV<String, TableRow>> element) {
       String key = element.getValue().getKey();
-      String tableName = String.format("%s:%s.%s", projectId, datasetName.get(), key);
+      String tableName = String.format("%s:%s.%s", projectId, datasetName, key);
       LOG.debug("Table Name {}", tableName);
       return KV.of(tableName, element.getValue().getValue());
     }

gcs-s3-inspect-config.json

@@ -34,7 +34,7 @@ public static void formatOptions(S3ImportOptions options) {
 
     options.setClientConfiguration(configuration);
 
-    if (options.getS3BucketUrl().get().toLowerCase().startsWith(AWS_S3_PREFIX)) {
+    if (options.getS3BucketUrl().toLowerCase().startsWith(AWS_S3_PREFIX)) {
       setAwsCredentials(options);
     }
 
@@ -46,8 +46,7 @@ public static void formatOptions(S3ImportOptions options) {
   private static void setAwsCredentials(S3ImportOptions options) {
     options.setAwsCredentialsProvider(
         new AWSStaticCredentialsProvider(
-            new BasicAWSCredentials(
-                options.getAwsAccessKey().get(), options.getAwsSecretKey().get())));
+            new BasicAWSCredentials(options.getAwsAccessKey(), options.getAwsSecretKey())));
   }
 
   private static void setAwsDefaultRegion(S3ImportOptions options) {