How to Setup Impression Integration To AWS S3 using AWS Kinesis Firehose and Lambda Cloud Services

This article shows step by step how to setup all the AWS cloud serverless components to store the Split impressions data in AWS S3 in JSON format

Split Admin site provide a webhook integration option to send the Impression data as they are injected into Split cloud, we will use the webhook to extract Impression data.

We will also use AWS API Gateway as the Webhook service, the API Gateway will forward the post request to AWS Lambda, which will extract the Impression information and send it to AWS Kinesis Firehose which has a built-in integration with S3.

Here are the steps to setup the integration:

1. First step is to create the Kinesis Firehose service, login to AWS and select Services->Kinesis->Data Firehose.

2. Put the service name, make sure Direct PUT or other sources option is set and click Next.

3. Click Next again into the Select a destination page, make sure S3 is selected, specify S3 bucket for the target storage, and click Next.

4. Select the IAM Role needed to access the service, consult with AWS expert for more info, in this example, we used a role that has AmazonKinesisFullAccess policy, click Next then click Create delivery stream.

5. The Second service needed is AWS Lambda, select AWS->Lambda, click on Create function button, choose Author from scratch option, use Java 8 for runtime engine.

6. Next step is to create the Java code for Lambda service, we will use Eclipse Maven project with the following dependencies:

<dependency>
   <groupId>com.amazonaws</groupId>
   <artifactId>aws-lambda-java-core</artifactId>
   <version>1.0.0</version>
</dependency>
<dependency>
   <groupId>com.amazonaws</groupId>
   <artifactId>aws-java-sdk-core</artifactId>
   <version>1.11.671</version>
</dependency>
<dependency>
    <groupId>com.amazonaws</groupId>
    <artifactId>amazon-kinesis-client</artifactId>
    <version>1.11.2</version>
 </dependency>
 <dependency>
    <groupId>org.json</groupId>
    <artifactId>json</artifactId>
    <version>20180813</version>
</dependency>

7. Insert the code below in the class (the class name here is WebhookConsumer.java), make sure to replace the awsAccessKey, awsSecretKey, fireHoseDeliveryStreamName and Regions.US-EAST-2. The code below will convert the JSON structure passed from the POST request to Dictionary objects, then loop through each Impression block to send it to Kinesis service.

package splitsupport;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.json.JSONObject;
import com.amazonaws.auth.AWSStaticCredentialsProvider;
import com.amazonaws.auth.BasicSessionCredentials;
import com.amazonaws.regions.Regions;
import com.amazonaws.services.kinesisfirehose.AmazonKinesisFirehose;
import com.amazonaws.services.kinesisfirehose.AmazonKinesisFirehoseClientBuilder;
import com.amazonaws.services.kinesisfirehose.model.PutRecordRequest;
import com.amazonaws.services.kinesisfirehose.model.PutRecordResult;
import com.amazonaws.services.kinesisfirehose.model.Record;
import com.amazonaws.services.lambda.runtime.Context;
import com.amazonaws.services.lambda.runtime.RequestHandler;

public class WebhookConsumer implements RequestHandler<Object, String>  {
@Override
 public String handleRequest(Object input, Context context) {
    String treatment="";
    try {
       String awsAccessKey="XXXXXXXX";
       String awsSecretKey="xxxxxxxxxxxxxxxxxxxxxx";
       String fireHoseDeliveryStreamName="SplitImpressions";
       BasicSessionCredentials awsCredentials = new BasicSessionCredentials(awsAccessKey, awsSecretKey);
       AmazonKinesisFirehose firehoseClient = AmazonKinesisFirehoseClientBuilder.standard()
          .withCredentials(new AWSStaticCredentialsProvider(awsCredentials))
          .withRegion(Regions.US_EAST_2)
          .build();
       List<Map<String , String>> impressions = StringToHash(input.toString());
       for (Map<String, String> imp1 : impressions) {
           JSONObject messageJson = new JSONObject();
           messageJson.put("key", imp1.get("key"));
           messageJson.put("split", imp1.get("split"));
           messageJson.put("environmentName", imp1.get("environmentName"));
           messageJson.put("time", imp1.get("time"));
           messageJson.put("label", imp1.get("label"));
           messageJson.put("treatment", imp1.get("treatment"));
           PutRecordRequest putRecordRequest = new PutRecordRequest();
           putRecordRequest.setDeliveryStreamName(fireHoseDeliveryStreamName);
           System.out.print(messageJson.toString());
           Record record = new Record().withData(ByteBuffer.wrap(messageJson.toString().getBytes()));
           putRecordRequest.setRecord(record);
           PutRecordResult putRecordResult = firehoseClient.putRecord(putRecordRequest);
           treatment="Success";              
       }
    } catch (Exception e) {
       System.out.print("Exception: "+e.getMessage());
       return "Exception: "+e.getMessage();
  }
  return treatment;
 }
 public static List<Map<String , String>> StringToHash(String decomp) {
     List<Map<String , String>> arrayRecord  = new ArrayList<Map<String,String>>();
     if (decomp.contains("}, {")) {
      for(final String entry : decomp.split("},")) {
       Map<String,String> oneRecord = new HashMap<String, String>();
        for(final String parts : entry.split(",")) {
         final String[] record = parts.split("=");
         String recordOne = record[0].replace("[{", "");
         recordOne = recordOne.replace("{", "");
         String recordTwo = record[1].replace("}]", "");
         oneRecord.put(recordOne.trim(), recordTwo.trim());
       }
       arrayRecord.add(oneRecord);
     }
   } else {   
     Map<String,String> oneRecord = new HashMap<String, String>();
     for(final String parts : decomp.split(",")) {
       final String[] record = parts.split("=");
       String recordOne = record[0].replace("[{", "");
       String recordTwo = record[1].replace("}]", "");
       oneRecord.put(recordOne.trim(), recordTwo.trim());
     }
     arrayRecord.add(oneRecord);
   }
   return arrayRecord;  
 }
}

8. Export a self-contained JAR file for your project, in  Eclipse, right-click on the project name and select Export.

9. Select Runnable JAR File option and click Next, then specify the JAR file name and click Finish.

10. Back to AWS Lambda page, under Function code section, update the Handler box with the content below:

splitsupport.WebhookConsumer::handleRequest

Then upload the new created JAR file.

11. The Third service needed is the API Gateway, select Services->API Gateway.

12. Click on Create API button, and use Rest API and New API options, type a name for your API service and click Create API.

13. Click on your new created API Service, click Actions list to add POST Method Execution, select Lambda Function for Integration type and type the Lambda function name created previously.

14. Since Split zip the post body when calling the Webhook, we need to enable binary type for our API Gateway, under Settings, add */* in Binary Media Types and click Save Changes.

15. The last step is to create a Stage for your API Gateway, and generate the webhook URL under Stages page, click the Create button.

16. Your Webhook service is ready now, login to the Split user interface, go to Admin Settings page, click on Integrations, and click Add button for Outgoing Webhook (Impressions) .

17. Select which environment's impressions to export, and paste the API Gateway URL, and click Save.