DLP Inspect initial commit #48
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -64,11 +64,12 @@ public PCollectionTuple expand(PCollection<KV<String, String>> input) { | |
| return input.apply( | ||
| "DLPInspect", | ||
| ParDo.of(new InspectData(projectId(), inspectTemplateName())) | ||
| .withOutputTags(Util.inspectData, TupleTagList.of(Util.auditData).and(Util.apiResponseFailedElements))); | ||
| } | ||
|
|
||
| public static class InspectData extends DoFn<KV<String, String>, Row> { | ||
| private String projectId; | ||
| private DlpServiceClient dlpServiceClient; | ||
| private String inspectTemplateName; | ||
| private InspectContentRequest.Builder requestBuilder; | ||
| private final Counter numberOfBytesInspected = | ||
|
|
@@ -87,9 +88,27 @@ public void setup() { | |
| .setInspectTemplateName(this.inspectTemplateName); | ||
| } | ||
|
|
||
| @StartBundle | ||
| public void startBundle() { | ||
| try { | ||
| this.dlpServiceClient = DlpServiceClient.create(); | ||
|
|
||
| } catch (IOException e) { | ||
| LOG.error("DLPTransform:DLPInspect: Failed to create DLP Service Client {}", e.getMessage()); | ||
| throw new RuntimeException(e); | ||
| } | ||
| } | ||
|
|
||
| @FinishBundle | ||
| public void finishBundle() { | ||
| if (this.dlpServiceClient != null) { | ||
| this.dlpServiceClient.close(); | ||
| } | ||
| } | ||
|
|
||
| @ProcessElement | ||
| public void processElement(ProcessContext c) throws IOException { | ||
| try { | ||
| String fileName = c.element().getKey(); | ||
|
|
||
| if (!c.element().getValue().isEmpty()) { | ||
|
|
@@ -99,7 +118,7 @@ public void processElement(ProcessContext c) throws IOException { | |
| if (this.requestBuilder.build().getSerializedSize() > DLP_PAYLOAD_LIMIT) { | ||
| String errorMessage = | ||
| String.format( | ||
| "DLPTransform:DLPInspect: Payload Size %s Exceeded Batch Size %s", | ||
|
There was a problem hiding this comment. Optional- Change the log.error to warn There was a problem hiding this comment. Yes got that. Will run this through the team and decide accordingly. |
||
| this.requestBuilder.build().getSerializedSize(), DLP_PAYLOAD_LIMIT); | ||
| LOG.error(errorMessage); | ||
| } else { | ||
|
|
@@ -109,7 +128,7 @@ public void processElement(ProcessContext c) throws IOException { | |
| long bytesInspected = contentItem.getSerializedSize(); | ||
| int totalFinding = | ||
| Long.valueOf(response.getResult().getFindingsList().stream().count()).intValue(); | ||
|
|
||
| boolean hasErrors = response.findInitializationErrors().stream().count() > 0; | ||
| if (response.hasResult() && !hasErrors) { | ||
| response | ||
|
|
@@ -128,7 +147,7 @@ public void processElement(ProcessContext c) throws IOException { | |
| finding.getLocation().getCodepointRange().getStart(), | ||
| finding.getLocation().getCodepointRange().getEnd()) | ||
| .build(); | ||
| LOG.debug("DLPTransform:DLPInspect: Row {}", row); | ||
|
|
||
| c.output(Util.inspectData, row); | ||
| }); | ||
|
|
@@ -148,7 +167,9 @@ public void processElement(ProcessContext c) throws IOException { | |
| Row.withSchema(Util.errorSchema) | ||
| .addValues(fileName, timeStamp, error.toString()) | ||
| .build()); | ||
| LOG.info("DLPTransform:DLPInspect: Initialization error in DLP response - {}", error); | ||
| }); | ||
| //Need to change 0 to 0L | ||
| c.output( | ||
| Util.auditData, | ||
| Row.withSchema(Util.bqAuditSchema) | ||
|
|
@@ -157,6 +178,17 @@ public void processElement(ProcessContext c) throws IOException { | |
| } | ||
| } | ||
| } | ||
| else{ | ||
| LOG.info("DLPTransform:DLPInspect: {} is an empty file | Size of the file in bytes - {}", fileName, c.element().getValue().length()); | ||
| c.output( | ||
| Util.auditData, | ||
| Row.withSchema(Util.bqAuditSchema) | ||
| .addValues(fileName, Util.getTimeStamp(),0L, "EMPTY") | ||
| .build()); | ||
| } | ||
|
There was a problem hiding this comment. For the processElement i think error is thrown and caught in the class after DLP is called. Can we have tuple like this? This should make sure internal error is not going to crash the pipeline. There was a problem hiding this comment. Yes I got it. But just for clarification the try block is already handling the exception, right? Also, this class is already returning a tuple. I will add this additional TupleTag for the Errors. There was a problem hiding this comment. Also should we handle this initialization error as follows : https://github.com/GoogleCloudPlatform/dlp-dataflow-deidentification/blob/2c9bbd178ccaaa4422f89ba49fd0e8b4a0e4f26b/src/main/java/com/google/swarm/tokenization/S3Import.java#L333 There was a problem hiding this comment. you can leave it. Ideally you will flatten all the errors and write back somewhere. processElement throws the error cand catch clock is to output without crashing the pipeline. |
||
| } | ||
| catch (Exception e) { | ||
| c.output(Util.apiResponseFailedElements, e.toString()); | ||
| } | ||
| } | ||
| } | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I think this is AuditInspectDataTransform
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Yes the class name is AuditInspectDataTransform. But for logging I have maintained the usage of Transformation names that will be displayed on UI DAG. So here the name of this Transformation is FileTrackerTransform and the sub step is MergePartialStatsRow. Can take out MergePartialStatsRow if it doesn't seem to be right as a log message.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
ok makes sense.