firehose_to_elasticsearch_with_slack_notify_on_error

index.js

const https = require('https');

const postSlack = (rec)=>{
  return new Promise((resolve, reject)=>{
    const path = rec.ecs_cluster.match(/staging/) ? process.env.STAGING_SLACK_PATH : process.env.PRODUCTION_SLACK_PATH;
    const data = JSON.stringify({ text: JSON.stringify(rec) });
    const options = {
      hostname: 'hooks.slack.com',
      port: 443,
      path,
      method: 'POST',
      headers: {
        'Content-Length': Buffer.byteLength(data),
        'User-Agent': "firehose-lambda",
        'Content-type': 'application/json'
      },
    };
    const req = https.request(options, (res) => {
      resolve();
    });
    req.on('error', (e) => {
      reject(e.message);
    });
    req.write(data);
    req.end();
  });
}

exports.handler = async (event, context) => {
  const errors = []
  const output = event.records.map((record) => {
    const buffer = Buffer.from(record.data, 'base64');
    const rec = JSON.parse(buffer)
    let message = ""
    if(rec.log){
      console.log(rec.log);
      return {
        recordId: record.recordId,
        result: 'Dropped'
      }
    }
    if(rec.status && rec.status >= 500){
      errors.push(rec);
    }
    message = JSON.stringify(rec);
    return {
      recordId: record.recordId,
      result: 'Ok',
      data: Buffer.from(message).toString('base64')
    }
  });
  if(errors.length === 0){
    return { records: output };
  }
  return new Promise((resolve, reject)=>{
    Promise.all(errors.map((e)=> postSlack(e))).finally(()=>{
      return resolve({ records: output });
    });
  });
};