Rewrite Cloudflare Log JSON before sending it to Coralogix

index.js

/**
 * AWS S3 Lambda function for Coralogix
 *
 * @file        https://gist.github.com/trieloff/1219ec5ec2f3ff42cad4b3f3ced5a429
 * @author      Coralogix Ltd. <info@coralogix.com>
 * @link        https://coralogix.com/
 * @copyright   Coralogix Ltd.
 * @licence     Apache-2.0
 * @version     1.0.8
 * @since       1.0.0
 */

"use strict";

// Import required libraries
const aws = require("aws-sdk");
const zlib = require("zlib");
const assert = require("assert");
const coralogix = require("coralogix-logger");
const s3 = new aws.S3();

// Check Lambda function parameters
assert(process.env.private_key, "No private key!");
const newlinePattern = process.env.newline_pattern ? RegExp(process.env.newline_pattern) : /(?:\r\n|\r|\n)/g;
const sampling = process.env.sampling ? parseInt(process.env.sampling) : 1;
const debug = JSON.parse(process.env.debug || false);

// Initialize new Coralogix logger
coralogix.CoralogixCentralLogger.configure(new coralogix.LoggerConfig({
    privateKey: process.env.private_key,
    debug: debug
}));
const logger = new coralogix.CoralogixCentralLogger();

/**
 * @description Send logs records to Coralogix
 * @param {Buffer} content - Logs records data
 * @param {string} filename - Logs filename S3 path
 */
function sendLogs(content, filename) {
    const logs = content.toString("utf8").split(newlinePattern);

    for (let i = 0; i < logs.length; i += sampling) {
        if (!logs[i]) continue;
        let appName = process.env.app_name || "NO_APPLICATION";
        let subName = process.env.sub_name || "NO_SUBSYSTEM";

        try {
            appName = appName.startsWith("$.") ? dig(appName, JSON.parse(logs[i])) : appName;
            subName = subName.startsWith("$.") ? dig(subName, JSON.parse(logs[i])) : subName;
        } catch {}

        logger.addLog(
            appName,
            subName,
            new coralogix.Log({
                severity: getSeverityLevel(logs[i]),
                text: jiggle(logs[i]),
                threadId: filename
            })
        );
    }
}

function jiggle(line) {
  const data = JSON.parse(line);
  return JSON.stringify({ cdn: {
    url: data.ClientRequestScheme + "://" + data.ClientRequestHost + data.ClientRequestURI,
    service_id: data.ZoneID,
    client: {
      ip: data.ClientIP,
      number: data.ClientASN,
      country_name: data.ClientCountry,
    },
    request: {
      user_agent: data.ClientRequestUserAgent,
      referer: data.ClientRequestReferer,
      protocol: data.ClientRequestProtocol,
      method: data.ClientRequestMethod,
      url: data.ClientRequestURI,
      id: data.RayID,
      accept_content: data.RequestHeaders.accept,
      accept_language: data.RequestHeaders.accept_language,
      accept_encoding: data.RequestHeaders.Accept_encoding,
      accept_charset: data.RequestHeaders.Accept_charset,
      connection: data.RequestHeaders.connection,
      forwarded: data.RequestHeaders.forwarded,
      via: data.RequestHeaders.via,
      xfh: data.RequestHeaders.x_forwarded_host,
      cache_control: data.RequestHeaders.cache_control,
    },
    time: {
      start: data.EdgeStartTimestamp,
      end: data.EdgeEndTimestamp,
      elapsed: data.EdgeTimeToFirstByteMs
    },
    helix: {
        type: "Cloudflare-" + data.ClientRequestSource,
    },
    response: {
      body_size: data.EdgeResponseBodyBytes,
      header_size: data.EdgeResponseBytes - data.EdgeResponseBodyBytes,
      content_type: data.EdgeResponseContentType,
      status: data.EdgeResponseStatus,
      error: data.ResponseHeaders.x_error,
      content_type: data.ResponseHeaders.content_type,
      age: data.ResponseHeaders.age,
      cache_control: data.ResponseHeaders.cache_control,
      expires: data.ResponseHeaders.expires,
      last_modified: data.ResponseHeaders.last_modified,
      vary: data.ResponseHeaders.vary,
    },
    edge: {
      datacenter: data.EdgeColoCode,
      ip: data.EdgeServerIP,
      cache_status: data.CacheCacheStatus,
    },
  }, cf: data });
}

/**
 * @description Extract nested field from object
 * @param {string} path - Path to field
 * @param {*} object - JavaScript object
 * @returns {*} Field value
 */
function dig(path, object) {
    if (path.startsWith("$.")) {
        return path.split(".").slice(1).reduce((xs, x) => (xs && xs[x]) ? xs[x] : path, object);
    }
    return path;
}

/**
 * @description Extract serverity from log record
 * @param {string} message - Log message
 * @returns {number} Severity level
 */
function getSeverityLevel(message) {
    const status = JSON.parse(message).EdgeResponseStatus;
    let severity = 3;
    if (message.includes("debug"))
        severity = 1;
    if (message.includes("verbose"))
        severity = 2;
    if (message.includes("info"))
        severity = 3;
    if (message.includes("warn") || message.includes("warning") || status >= 400)
        severity = 4;
    if (message.includes("error") || status >= 500)
        severity = 5;
    if (message.includes("critical") || message.includes("panic"))
        severity = 6;
    return severity;
}

/**
 * @description Lambda function handler
 * @param {object} event - Event data
 * @param {object} context - Function context
 * @param {function} callback - Function callback
 */
function handler(event, context, callback) {
    const bucket = event.Records[0].s3.bucket.name;
    const key = decodeURIComponent(event.Records[0].s3.object.key.replace(/\+/g, " "));

    s3.getObject({
        Bucket: bucket,
        Key: key
    }, (error, data) => {
        if (error) {
            callback(error);
        } else {
            if (data.ContentType == "application/x-gzip" ||
                data.ContentEncoding == "gzip" ||
                data.ContentEncoding == "compress" ||
                key.endsWith(".gz")
            ) {
                zlib.gunzip(data.Body, (error, result) => {
                    if (error) {
                        callback(error);
                    } else {
                        sendLogs(Buffer.from(result));
                        callback(null, data.ContentType);
                    }
                });
            } else {
                sendLogs(Buffer.from(data.Body), `s3://${bucket}/${key}`);
            }
        }
    });
}

exports.handler = handler;