feat: transformer: sideline partially erroring lines (#121)

matanolabs · Apr 11, 2023 · 682376e · 682376e
1 parent c6ec2ff
commit 682376e
Show file tree

Hide file tree

Showing 10 changed files with 285 additions and 35 deletions.
diff --git a/infra/bin/app.ts b/infra/bin/app.ts
@@ -27,6 +27,7 @@ const dpMainStack = new DPMainStack(app, "DPMainStack", {
   realtimeBucket: dpCommonStack.realtimeBucket,
   realtimeBucketTopic: dpCommonStack.realtimeBucketTopic,
   matanoAthenaResultsBucket: dpCommonStack.matanoAthenaResultsBucket,
+  transformerSidelineBucket: dpCommonStack.transformerSidelineBucket,
   integrationsStore: dpCommonStack.integrationsStore,
   alertTrackerTable: dpCommonStack.alertTrackerTable,
 });

diff --git a/infra/lib/transformer.ts b/infra/lib/transformer.ts
@@ -8,6 +8,7 @@ import * as lambda from "aws-cdk-lib/aws-lambda";
 import { RustFunctionLayer } from "./rust-function-layer";
 
 interface TransformerProps {
+  sidelineBucket: s3.IBucket;
   realtimeBucket: s3.IBucket;
   realtimeTopic: sns.Topic;
   matanoSourcesBucket: s3.IBucket;
@@ -39,6 +40,7 @@ export class Transformer extends Construct {
         MATANO_REALTIME_BUCKET_NAME: props.realtimeBucket.bucketName,
         MATANO_REALTIME_TOPIC_ARN: props.realtimeTopic.topicArn,
         SQS_METADATA: props.sqsMetadata,
+        MATANO_SIDELINE_BUCKET: props.sidelineBucket.bucketName,
       },
       layers: [this.rustFunctionLayer.layer],
       timeout: cdk.Duration.seconds(100),
@@ -56,6 +58,7 @@ export class Transformer extends Construct {
       ],
     });
 
+    props.sidelineBucket.grantReadWrite(this.transformerLambda);
     props.matanoSourcesBucket.grantRead(this.transformerLambda);
     props.realtimeBucket.grantWrite(this.transformerLambda);
     props.realtimeTopic.grantPublish(this.transformerLambda);

diff --git a/infra/src/DPCommonStack.ts b/infra/src/DPCommonStack.ts
@@ -23,6 +23,7 @@ export class DPCommonStack extends MatanoStack {
   integrationsStore: IntegrationsStore;
   alertTrackerTable: ddb.Table;
   matanoAthenaResultsBucket: s3.Bucket;
+  transformerSidelineBucket: s3.Bucket;
 
   constructor(scope: Construct, id: string, props: DPCommonStackProps) {
     super(scope, id, props);
@@ -135,6 +136,8 @@ export class DPCommonStack extends MatanoStack {
       this.integrationsStore = new IntegrationsStore(this, "MatanoIntegrationsStore", {});
     }
 
+    this.transformerSidelineBucket = new Bucket(this, "MatanoTransformerSidelineBucket");
+
     this.humanCfnOutput("MatanoIngestionS3BucketName", {
       value: this.matanoIngestionBucket.bucket.bucketName,
       description:
@@ -147,8 +150,14 @@ export class DPCommonStack extends MatanoStack {
         "The name of the S3 Bucket used for long term storage backing your data lake. See https://www.matano.dev/docs/tables/querying",
     });
 
+    this.humanCfnOutput("MatanoTransformerSidelineS3BucketName", {
+      value: this.transformerSidelineBucket.bucketName,
+      description: "The name of the S3 Bucket where erroring lines are sidelined.",
+    });
+
     // important: to prevent output deletion
     this.exportValue(this.matanoIngestionBucket.topic.topicArn);
+    this.exportValue(this.transformerSidelineBucket.bucketArn);
     this.exportValue(this.matanoIngestionBucket.bucket.bucketArn);
     this.exportValue(this.matanoAthenaResultsBucket.bucketArn);
   }

diff --git a/infra/src/DPMainStack.ts b/infra/src/DPMainStack.ts
@@ -51,6 +51,7 @@ interface DPMainStackProps extends MatanoStackProps {
   realtimeBucket: Bucket;
   realtimeBucketTopic: sns.Topic;
   matanoAthenaResultsBucket: s3.Bucket;
+  transformerSidelineBucket: s3.Bucket;
   integrationsStore: IntegrationsStore;
   alertTrackerTable: Table;
 }
@@ -194,6 +195,7 @@ export class DPMainStack extends MatanoStack {
       realtimeBucket: props.realtimeBucket,
       realtimeTopic: props.realtimeBucketTopic,
       matanoSourcesBucket: props.matanoSourcesBucket.bucket,
+      sidelineBucket: props.transformerSidelineBucket,
       logSourcesConfigurationPath: path.join(this.configTempDir, "config"), // TODO: weird fix later (@shaeq)
       sqsMetadata: sqsSources.sqsMetadata,
     });

diff --git a/lib/rust/Cargo.lock b/lib/rust/Cargo.lock
diff --git a/lib/rust/lake_writer/Cargo.toml b/lib/rust/lake_writer/Cargo.toml
@@ -8,7 +8,7 @@ edition = "2021"
 [dependencies]
 shared = { path = "../shared", features = ["avro"] }
 rayon = "1.5.3"
-zstd = "0.11.2"
+zstd = "0.12.1"
 base64 = "0.13.1"
 bytes = "1.2.1"
 tokio = { version = "1.17.0", features = ["full"] }

diff --git a/lib/rust/log_puller/Cargo.toml b/lib/rust/log_puller/Cargo.toml
@@ -40,7 +40,7 @@ async-trait = "0.1.58"
 enum_dispatch = "0.3.8"
 regex = "1"
 async-stream = "0.3.3"
-zstd = "0.11.2"
+zstd = "0.12.1"
 walkdir = "2.3.2"
 zip = "0.6.3"
 config = { version = "0.13.1", features = ["yaml"] }

diff --git a/lib/rust/shared/src/utils.rs b/lib/rust/shared/src/utils.rs
@@ -228,3 +228,11 @@ pub fn convert_json_array_str_to_ndjson(s: &str) -> Result<String> {
 
     Ok(res)
 }
+
+pub fn result_to_option<T, E>(r: Result<Option<T>, E>) -> Option<Result<T, E>> {
+    match r {
+        Ok(Some(v)) => Some(Ok(v)),
+        Ok(None) => None,
+        Err(e) => Some(Err(e)),
+    }
+}
diff --git a/lib/rust/transformer/Cargo.toml b/lib/rust/transformer/Cargo.toml
@@ -20,6 +20,7 @@ async-compression = { version = "0.3.14", default-features = false, features = [
   "zstd",
   "stream",
 ] }
+zstd = "0.12.1"
 infer = "0.12.0"
 regex = "1.5.4"
 arrow2 = { git = "https://github.com/jorgecarleitao/arrow2", features = [