Ingesting TV Response Data into Your Own S3 Bucket

This article contains the following:

• Overview
• Ingestion setup steps 
  • Credentials
  • Scripts 

Overview

Tatari offers access to granular TV response data via Tatari S3 buckets. Once you have access to our S3 bucket, there are many ways to automate ingestion of this data into your data warehouse or internal tools. 

This article will help you automate ingestion of TV response data from Tatari S3 buckets to your own buckets. Note: This guide is in beta so please let us know if you run into any issue at client-integrations@tatari.tv. 

Ingestion Setup Steps

Before ingesting TV response data into your own S3 bucket, you have to have set up access to your reports in Tatari's S3 bucket first. Follow the instructions in your dashboard’s AWS Data Sharing page to do so.

1. Log in to your dashboard
2. On the bottom left navigation bar, select “Settings” -> “Integrations”
3. Select "AWS Data Sharing" from the top navigation bar.

 Then, follow the remaining steps below:

1. Use your account’s AWS credentials to fetch temporary credentials to access your Tatari S3 folder.
2. Pull file lists from both Tatari’s report folder and your destination bucket, looking for files that don’t exist in the latter.
3. Use the AWS library to download and then upload the contents of each file.

That’s it! This script is also built to be immediately usable as a Lambda function.

This script will sync your Tatari reports folder with a bucket under your control. A report at

s3://tatari-reports/exports/<your_company>/<report>/v2/<report>_<date_range>.csv.gz 

will be copied to

s3://<your_bucket>/<your_company>/<report>/v2/<report>_<date_range>.csv.gz

Credentials

You will need the following:

• Your Tatari slug
• Your AWS Access Key
• Your AWS Secret Key

Scripts

Currently, only a NodeJS script is available. If you’re interested in another language, please let us know at client-integrations@tatari.tv. 

NodeJS

The Node script as-is expects your credentials and configuration to be hard-coded.

Requirements

yarn add aws-sdk

Script

const AWS = require('aws-sdk');

const _ = {}

// Semantic Version
// Created 2020-09-02
_.VERSION = '0.0.1'


// Credentials
_.accessKeyId = 'YOUR_KEY';
_.secretAccessKey = 'YOUR_SECRET';


// Configuration
_.tatariSlug = "YOUR_SLUG";
_.destinationBucket = 'YOUR_BUCKET'


// Constants
_.sourceBucket = "tatari-reports-exports";
_.role = `arn:aws:iam::878256633362:role/ClientRoles/DataShare/ClientRole-DataShare-${_.tatariSlug}`


_.createClientS3Client = function (key, secret) {
  return new AWS.S3({
    accessKeyId: key,
    secretAccessKey: secret
  })
}


_.assumeRoleCredentials = async function (stsClient, role) {
  const timestamp = (new Date()).getTime();
  const sts_params = {
      RoleArn: role,
      RoleSessionName: `id-${timestamp}`,
      DurationSeconds: 3600
  };
  const { Credentials } = await stsClient.assumeRole(sts_params).promise();
  return Credentials
}


_.createTatariS3Client = async function (key, secret) {
  const sts = new AWS.STS({ accessKeyId: key, secretAccessKey: secret});

  const { AccessKeyId, SecretAccessKey, SessionToken } = await _.assumeRoleCredentials(sts, _.role);

  return new AWS.S3({
    accessKeyId: AccessKeyId,
    secretAccessKey: SecretAccessKey,
    sessionToken: SessionToken
  });
}


_.isAFile = function (objectKey) {
  return objectKey.indexOf('.') > -1
}


_.fetchObjectKeys = async function (client, bucket, prefix) {
  const files = []
  const params = { Bucket: bucket, Prefix: prefix + "/" }

  for (;;) {
      let data = await client.listObjectsV2(params).promise();
      data.Contents.forEach(elem => files.push(elem.Key));

      if (!data.IsTruncated) { break; }
      params.ContinuationToken  = data.NextContinuationToken;
  }

  return files
}


_.errorCounter = function (status) {
  return (err, data) => {
    if (err) {
        status.fail++;
    } else {
        status.success++;
    }
  }
}


_.copyTo = function (client, bucket, status) {
  return (data, Key) => {
    const params = {
        Bucket: bucket,
        Key: Key,
        Body: Buffer.from(data.Body, 'base64')
    };

    client.upload(params, _.errorCounter(status));
  }
}


_.downloadFrom = function (client, bucket, status, callback) {
  return async function(Key) {
    await client.getObject({
        Bucket: bucket,
        Key: Key
    }, (err, data) => {
        if (err) {
          status.fail++;
        } else {
          callback(data, Key)
        }
    });
  }
}


_.wait = function () {
    return new Promise((resolve, reject) => {
        setTimeout(() => resolve("waiting..."), 1000)
    });
}


_.sync = async function (key, secret, source, prefix, destination) {
  const destinationS3 = _.createClientS3Client(key, secret);
  const tatariS3 = await _.createTatariS3Client(key, secret);

  const destinationKeys = await _.fetchObjectKeys(destinationS3, destination, _.tatariSlug)
  const destinationFiles = destinationKeys.filter(_.isAFile);

  const tatariKeys = await _.fetchObjectKeys(tatariS3, source, prefix)
  const filesToCopy = tatariKeys.filter(_.isAFile).filter(x => !destinationFiles.includes(x))

  const status = {
    total: filesToCopy.length,
    success: 0,
    fail: 0,
  }

  const duplicate = _.downloadFrom(
    tatariS3, source, status, 
    _.copyTo(destinationS3, destination, status)
  )

  await Promise.all(filesToCopy.map(duplicate));

  while (status.fail + status.success < status.total) { await _.wait(); }

  return status

}


module.exports = {
  config: _,
  Sync: () => _.sync(
    _.accessKeyId,
    _.secretAccessKey,
    _.sourceBucket,
    _.tatariSlug,
    _.destinationBucket,
  ),

  handler: (callEvent, context, callback) => {
    _.sync(
      _.accessKeyId,
      _.secretAccessKey,
      _.sourceBucket,
      _.tatariSlug,
      _.destinationBucket,
    ).then(r => {
      context.succeed({success: true, status: status});
      callback(null, {})
    })
  }
}

Lambda

This script can be dropped into a Lambda function once you have replaced the placeholder values for your AWS credentials, destination bucket, and company slug. You will need to increase the Lambda function timeout to the maximum of 15 minutes and set up a method to trigger that lambda function (such as a cloudwatch alarm).