Add Kinesis Stream Metadata to firehose event (#188)

* Add source kinesis stream arn to firehose event

* Add Kinesis Record Metadata struct to allow for reingest to a stream

* fix approximateArrivalTimestamp in test data

updated type to match sample in Lambda console

Author: bhavikkumar

events/firehose.go

@@ -4,16 +4,18 @@ package events
 
 // KinesisFirehoseEvent represents the input event from Amazon Kinesis Firehose. It is used as the input parameter.
 type KinesisFirehoseEvent struct {
-	InvocationID      string                       `json:"invocationId"`
-	DeliveryStreamArn string                       `json:"deliveryStreamArn"`
-	Region            string                       `json:"region"`
-	Records           []KinesisFirehoseEventRecord `json:"records"`
+	InvocationID           string                       `json:"invocationId"`
+	DeliveryStreamArn      string                       `json:"deliveryStreamArn"`
+	SourceKinesisStreamArn string                       `json:"sourceKinesisStreamArn"`
+	Region                 string                       `json:"region"`
+	Records                []KinesisFirehoseEventRecord `json:"records"`
 }
 
 type KinesisFirehoseEventRecord struct {
-	RecordID                    string                `json:"recordId"`
-	ApproximateArrivalTimestamp MilliSecondsEpochTime `json:"approximateArrivalTimestamp"`
-	Data                        []byte                `json:"data"`
+	RecordID                      string                        `json:"recordId"`
+	ApproximateArrivalTimestamp   MilliSecondsEpochTime         `json:"approximateArrivalTimestamp"`
+	Data                          []byte                        `json:"data"`
+	KinesisFirehoseRecordMetadata KinesisFirehoseRecordMetadata `json:"kinesisRecordMetadata"`
 }
 
 // Constants used for describing the transformation result
@@ -32,3 +34,10 @@ type KinesisFirehoseResponseRecord struct {
 	Result   string `json:"result"` // The status of the transformation. May be TransformedStateOk, TransformedStateDropped or TransformedStateProcessingFailed
 	Data     []byte `json:"data"`
 }
+
+type KinesisFirehoseRecordMetadata struct {
+	ShardID                     string                `json:"shardId"`
+	PartitionKey                string                `json:"partitionKey"`
+	SequenceNumber              string                `json:"sequenceNumber"`
+	ApproximateArrivalTimestamp MilliSecondsEpochTime `json:"approximateArrivalTimestamp"`
+}

events/testdata/kinesis-firehose-event.json

@@ -1,6 +1,7 @@
 {
    "invocationId": "invoked123",
    "deliveryStreamArn": "aws:lambda:events",
+   "sourceKinesisStreamArn": "arn:aws:kinesis:us-east-1:123456789012:stream/test",
    "region": "us-west-2",
    "records": [
      {
@@ -10,7 +11,7 @@
        "kinesisRecordMetadata": {
          "shardId": "shardId-000000000000",
          "partitionKey": "4d1ad2b9-24f8-4b9d-a088-76e9947c317a",
-         "approximateArrivalTimestamp": "1507217624302",
+         "approximateArrivalTimestamp": 1507217624302,
          "sequenceNumber": "49546986683135544286507457936321625675700192471156785154",
          "subsequenceNumber": ""
        }
@@ -22,10 +23,10 @@
        "kinesisRecordMetadata": {
          "shardId": "shardId-000000000001",
          "partitionKey": "4d1ad2b9-24f8-4b9d-a088-76e9947c318a",
-         "approximateArrivalTimestamp": "1507217624302",
+         "approximateArrivalTimestamp": 1507217624302,
          "sequenceNumber": "49546986683135544286507457936321625675700192471156785155",
          "subsequenceNumber": ""
        }
      }
    ]
- }
+ }