@opentribe/packages_logger/analytics/index.js

const fs_extra = require('fs-extra');
const fs = require('fs');
const path = require('path');
const fetch = require("node-fetch");
const shell = require("shelljs");
const line_by_line = require('line-by-line');
const async_series = require('async/series');
const zlib = require("zlib");
const jsome = require('jsome');

const Promises = require('@opentribe/packages_promises');

const LogInsights = require('./insights.js');
const LogAdapters = require('./adapters.js');

const DataDir = path.resolve(__dirname, './data');
const ArchiveDir = path.resolve(__dirname, './s3-log-archive');
const BucketName = "open-tribe-logdna-archive";

const moveFolders = (oldPath, newPath, callback) => {
  fs.rename(oldPath, newPath, function (err) {
    if (err) {
      if (err.code === 'EXDEV') {
        return copy();
      }
      return callback(err);
    }
    callback();
  });
  const copy = () => {
    var readStream = fs.createReadStream(oldPath);
    var writeStream = fs.createWriteStream(newPath);
    readStream.on('error', callback);
    writeStream.on('error', callback);
    readStream.on('close', () => fs.unlink(oldPath, callback));
    readStream.pipe(writeStream);
  }
};

const safe_fetch = (...args) => new Promise(async (resolve, reject) => { try {
  const response = await fetch(...args);
  if (response.ok) {
    const text = await response.text();
    resolve(text)
  }
  reject(response);
} catch(err) { reject(err) } });

const get_last_5000_logs = (service_key, target) => new Promise(async (resolve, reject) => { try {
  const text = await safe_fetch(`https://user:${service_key}@api.logdna.com/v1/export?to=${Date.now()}&from=${0}`);
  fs.writeFileSync(target, text);
  resolve();
} catch(err) { reject(err) } });

const sync_archived_logs = () => new Promise(async (resolve, reject) => { try {
  const child = shell.exec('aws s3 sync s3://' + BucketName + ' ' + ArchiveDir + '/');
  resolve();
} catch(err) { reject(err) } });

const resync_archived_logs = () => new Promise(async (resolve, reject) => { try {
  fs_extra.removeSync(ArchiveDir);
  fs.mkdirSync(ArchiveDir);
  const child = shell.exec('aws s3 sync s3://' + BucketName + ' ' + ArchiveDir + '/');
  resolve();
} catch(err) { reject(err) } });

const segregate_relevant_logs = () => new Promise(async (resolve, reject) => { try {
  Object.keys(LogAdapters).map(type => {
    fs.writeFileSync(DataDir + '/' + type + '.txt', '');
  });
  const log_line_by_line = new line_by_line(DataDir + '/raw.txt');
  let log_stats = {};
  log_line_by_line.on('line', (line) => {
    if (line.length > 0) {
      let line_json = JSON.parse(line);
      Object.keys(LogAdapters)
        .filter(type => LogAdapters[type].condition(line_json))
        .map(type => ([type, LogAdapters[type].adapter(line_json)]))
        .forEach(([type, mapped_data]) => {
          if (typeof log_stats[type] === 'undefined') {
            log_stats[type] = 1;
          }
          log_stats[type] = log_stats[type] + 1;
          fs.appendFileSync(DataDir + '/' + type + '.txt', JSON.stringify(mapped_data) + '\n')
        })
    }
  })
  log_line_by_line.on('error', (err) => {
    console.log(err);
    reject(err);
  })
  log_line_by_line.on('end', () => {
    jsome({
      available_shards: log_stats
    });
    resolve();
  })
} catch(err) { reject(err) } });

const filepathList = (dir, filelist) => {
  files = fs.readdirSync(dir);
  filelist = filelist || [];
  files.forEach(file => {
    if (fs.statSync(dir + '/' + file).isDirectory()) {
      filelist = filepathList(dir + '/' + file, filelist);
      return
    }
    filelist.push(dir + '/' + file);
  });
  return filelist;
};

const unzip_merge_logs = (targetdir, outputfile) => new Promise(async (resolve, reject) => { try {
  fs.writeFileSync(outputfile, '');
  const filesToUnzip = filepathList(targetdir);
  await Promises.series(filesToUnzip.map(filePath => () => new Promise(async (resolve, reject) => { try {
    const _d = fs.readFileSync(filePath);
    zlib.gunzip(_d, (err, decoded) => {
      if (err) {
        console.log(err);
        reject(err)
      }
      console.log('Appending log chunk of size: ' + (fs.statSync(filePath).size / 1000000).toFixed(2) + ' MBs');
      fs.appendFileSync(outputfile, decoded.toString());
      resolve();
    });
  } catch(err) { reject(err) } })));
  resolve();
} catch(err) { reject(err) } });

exports.prepare_resync = () => new Promise(async (resolve, reject) => { try {
  await resync_archived_logs();
  await unzip_merge_logs(ArchiveDir, DataDir + '/raw.txt');
  await segregate_relevant_logs();
  resolve();
} catch(err) { reject(err) } });

exports.prepare_live = (service_key) => new Promise(async (resolve, reject) => { try {
  await get_last_5000_logs(service_key, DataDir + '/raw.txt');
  await segregate_relevant_logs();
  resolve();
} catch(err) { reject(err) } });

exports.prepare_archived = () => new Promise(async (resolve, reject) => { try {
  await sync_archived_logs();
  await unzip_merge_logs(ArchiveDir, DataDir + '/raw.txt');
  await segregate_relevant_logs();
  resolve();
} catch(err) { reject(err) } });

//good for iterating over an idea on how to map a particular log file
//without incurring a request limit penalty
exports.prepare_from_cache = () => new Promise(async (resolve, reject) => { try {
  await segregate_relevant_logs();
  resolve();
} catch(err) { reject(err) } });

exports.insight = (query_name, args) => LogInsights[query_name](...(args || []));
exports.insight_list = () => new Promise((resolve, reject) => {
  console.log(JSON.stringify(Object.keys(LogInsights), null, 2));
  resolve();
});