Using MongoDB Realm WebHooks with Amazon Kinesis Data Firehose

#Status Codes

The HTTP status code must be in the 2xx, 4xx, 5xx range; they will not follow redirects, so nothing in the 3xx range. Only a status of 200 is considered a successful delivery of the records; all other statuses are regarded as a retriable error, except 413.

413 (size exceeded) is considered a permanent failure, and will not be retried. In all other error cases, they will reattempt delivery of the same batch of records using an exponential back-off algorithm.

The retries are backed off using an initial back-off time of 1 second with a jitter factor of 15% . Each subsequent retry is backed off using the formula initial-backoff-time * (multiplier(2) ^ retry_count) with added jitter. The back-off time is capped by a maximum interval of 2 minutes. For example on the 'n'-th retry the back-off time is = MAX(120sec, (1 * (2^n)) * random(0.85, 1.15).

These parameters are subject to change. Please refer to the AWS Firehose documentation for exact initial back-off time, max back-off time, multiplier, and jitter percentages.

#Other Response Headers

As well as the HTTP status code your response should include the following headers:

• Content-Type: The only acceptable content type is application/json
• Content-Length: The Content-Length header must be present if the response has a body

Do not send a Content-Encoding header, the body must be uncompressed.

#Response Body

Just like the Request, the Response body is JSON, but it has a max filesize of 1MiB. This JSON body has two required properties:

• requestId: This must match the requestId in the Delivery Stream Request
• timestamp: The timestamp (milliseconds since epoch) at which the server processed this request

If there was a problem processing the request, you could optionally include an errorMessage property. If a request fails after exhausting all retries, the last Instance of this error message is copied to the error output S3 bucket, if one has been configured for the Delivery Stream.

#Decoding and Parsing the Payload Body

1
// Payload body is a JSON string, convert into a JavaScript Object
2
const data = JSON.parse(payload.body.text())
3
4
// Each record is a Base64 encoded JSON string
5
const documents = data.records.map((record) => {
6
    const document = JSON.parse(decodeBase64(record.data))
7
    return {
8
        ...document,
9
        _id: new BSON.ObjectId(document._id)
10
    }
11
})

When we receive the POST request, we first have to convert the body—which is a JSON string—into a JavaScript object. Then we can iterate over each of the records.

The data in each of these records is Base64 encoded, so we have to decode it first.

This data could be anything, whatever you've supplied in your Delivery Stream, but in this example, it is the MongoDB document sent from our Realm trigger. This document is also a JSON string, so we'll need to parse it back into a JavaScript object.

#Writing the Reports to MongoDB Atlas

Once the parsing and decoding are complete, we're left with an array of between 1 and 10,000 objects, depending on the size of the batch. It's tempting to pass this array to insertMany(), but there is the possibility that some records might already exist as documents in our collection.

Remember if Kinesis does not receive an HTTP status of 200 in response to a request it will, in the majority of cases, retry the batch. We have to take into account that there could be an issue after the documents have been written that prevents Kinesis from receiving the 200 OK status. If this occurs and we try to insert the document again, MongoDB will raise a Duplicate key error exception.

To prevent this we perform a find() and updateOne(), with upsert().

When updating/inserting a single document, you can use updateOne() with the upsert option.

1
context.services.get("mongodb-atlas").db("monitors").collection("firehose").updateOne(
2
    {_id: document._id},
3
    document,
4
    {upsert: true}
5
)

But we could potentially have to update/insert 10,000 records, so instead, we perform a bulk write.

1
// Perform operations as a bulk
2
const bulkOp = context.services.get("mongodb-atlas").db("monitors").collection("firehose").initializeOrderedBulkOp()
3
documents.forEach((document) => {
4
    bulkOp.find({ _id:document._id }).upsert().updateOne(document)
5
})

#Sending the Response

1
bulkOp.execute().then(() => {
2
    // All operations completed successfully
3
    response.setStatusCode(200)
4
    response.setBody(JSON.stringify({
5
        requestId: payload.headers['X-Amz-Firehose-Request-Id'][0],
6
        timestamp: (new Date()).getTime()
7
    }))
8
    return
9
})

If our write operations have completed successfully, we return an HTTP 200 status code with our response. Otherwise, we return a 500 and include the error message from the exception in the response body.

1
).catch((error) => {
2
    // Catch any error with execution and return a 500
3
    response.setStatusCode(500)
4
    response.setBody(JSON.stringify({
5
        requestId: payload.headers['X-Amz-Firehose-Request-Id'][0],
6
        timestamp: (new Date()).getTime(),
7
        errorMessage: error
8
    }))
9
    return
10
})