CR-5828 logs-offloader (#709)

rotem-codefresh · web-flow · commit 66708482802a · 2021-08-12T16:59:48.000+03:00
diff --git a/lib/interface/cli/commands/context/create/secret-store/base.cmd.js b/lib/interface/cli/commands/context/create/secret-store/base.cmd.js
@@ -1,3 +1,5 @@
+const yargs = require('yargs');
+
 const Command = require('../../../../Command');
 const createContext = require('../../create.cmd');
 
diff --git a/lib/interface/cli/commands/offline-logs/base.cmd.js b/lib/interface/cli/commands/offline-logs/base.cmd.js
@@ -0,0 +1,46 @@
+const yargs = require('yargs');
+const Command = require('../../Command');
+
+const command = new Command({
+    command: 'offline-logs',
+    root: true,
+    description: 'Manages offline logs',
+    webDocs: {
+        category: 'Logs category',
+        subCategory: 'Logs sub category',
+        title: 'Archives old logs to file or collection.',
+    },
+    builder: (yargs) => {
+        // set options which are used in both sub-commands
+        return yargs
+        .option('uri', {
+              describe: "Mongodb URI",
+              demandOption: true,
+              type: "string",
+            })
+        .option('db', {
+          describe: "Database name",
+          demandOption: true,
+          type: "string",
+            })
+        .option('collections', {
+          alias: "c",
+          describe: "Source collections names",
+          default: ["logs", "metadata"],
+          array: true,
+          type: "string",
+        })
+        .option('cutoffDate', {
+          alias: "cod",
+          describe:
+            "The date in ISO format (yyyy-MM-dd) from which logs will be archived (going backwards, including logs from that day).",
+          demandOption: true,
+          type: "string",
+        });
+    },
+    handler: () => {
+        yargs.showHelp();
+    },
+});
+
+module.exports = command;
diff --git a/lib/interface/cli/commands/offline-logs/offload-to-collection.cmd.js b/lib/interface/cli/commands/offline-logs/offload-to-collection.cmd.js
@@ -0,0 +1,88 @@
+const { MongoClient, ObjectId } = require("mongodb");
+const moment = require('moment')
+const Command = require('../../Command');
+const cmd = require('./base.cmd');
+const { objectIdFromDate } = require('./utils')
+
+const offloadToCollection = async function(sourceDBObj, collection, targetDB, cutoffDate) {
+  const sourceCollectionObj = sourceDBObj.collection(collection)
+  const targetCollection = `archive-${collection}`
+  
+  const cutoffDateObj = moment(cutoffDate)
+                        .add(1, 'days')
+                        .startOf("day").toDate()
+
+  if(!cutoffDateObj.isValid()){
+    throw new Error('please enter a valid date in ISO 8601 format')
+  }
+
+  const cutoffDateId = objectIdFromDate(cutoffDateObj)
+
+  var result = sourceCollectionObj.aggregate([
+      { $match: { _id: { $lte: ObjectId(cutoffDateId) } } },
+      { $merge: {
+          into: {db: targetDB, coll: targetCollection},
+          on: "_id",
+          whenMatched: "keepExisting",
+          whenNotMatched: "insert"
+      }}
+  ])
+
+  await result.toArray()
+
+  if (!result.cursorState.killed){
+    await sourceCollectionObj.deleteMany({_id: {$lte: ObjectId(cutoffDateId)}})
+  }
+  else {
+      console.error("Cursor error. Archiving operation may not be completed.")
+      console.error("The old logs were not deleted from the source collection.")
+  }
+}
+
+const command = new Command({
+    command: 'offload-to-collection',
+    parent: cmd,
+    description: 'Archiving logs from one or more source collections to target collections.',
+    webDocs: {
+        category: 'Logs',
+        title: 'Offload To Collection',
+    },
+    builder: yargs => yargs
+    .option('targetDB', {
+          alias: 'tdb',
+          describe: "Target database name, if none inserted, db will be defined as target.",
+          type: "string",
+        })
+        .example('codefresh offline-logs offload-to-collection --uri "mongodb://192.168.99.100:27017" --db logs --c logs foo --cod "2021-07-08" '),
+    handler: async (argv) => {
+      const {
+        uri,
+        db,
+        collections,
+        targetDB,
+        cutoffDate,
+      } = argv
+      const client = new MongoClient(uri);
+      try{
+        await client.connect()
+        const failedCollections = [];
+        const sourceDBObj = client.db(db);
+        const promises = collections.map( async (collection) => {
+          try{
+            await offloadToCollection(sourceDBObj, collection, targetDB || db, cutoffDate);
+          } catch (error) {
+            failedCollections.push(collection)
+          }
+        })
+        await Promise.all(promises)
+
+        if (failedCollections.length){
+          throw new Error(`failed to offload from collections: ${failedCollections.join(', ')}`)
+        }
+      } finally {
+          client.close();
+      }
+    },
+})
+
+module.exports = command;
diff --git a/lib/interface/cli/commands/offline-logs/offload-to-file.cmd.js b/lib/interface/cli/commands/offline-logs/offload-to-file.cmd.js
@@ -0,0 +1,135 @@
+
+const { MongoClient, ObjectId } = require("mongodb");
+const moment = require('moment')
+const Command = require('../../Command');
+const cmd = require('./base.cmd');
+const {objectIdFromDate, writeLogsToFile, checkRemnant} = require('./utils')
+const { boolean } = require("yargs");
+
+const offloadToFile = async function(database, collection, chunkDuration, cutoffDate, path) {
+  const collectionObj = database.collection(collection);
+
+  if (chunkDuration <= 0){
+    throw new Error('please enter a valid chunkDuration ( > 0)')
+  }
+  let chunkSize = chunkDuration
+
+  // the cutoff date will be the start of the next day from user input (archive will include the given cutoff date)
+  const cutoffDateObj = moment(cutoffDate, moment.ISO_8601, true)
+    .add(1, 'days')
+    .startOf("day");
+  if(!cutoffDateObj.isValid()){
+    throw new Error('please enter a valid date in ISO 8601 format')
+  }
+  const cutoffDateId = objectIdFromDate(cutoffDateObj.toDate())
+
+  const minLog = await collectionObj
+    .find({ _id: { $lt: ObjectId(cutoffDateId) } })
+    .sort({ _id: 1 })
+    .limit(1)
+    .toArray();
+  
+  if (minLog.length === 0) {
+      console.info(`No logs to archive in Collection ${collection} from the given date.`);
+      return;
+  }
+  
+  const minLogId = ObjectId(minLog[0]._id)
+  const minDate = moment(
+    minLogId.getTimestamp()
+  ).startOf("day").toDate();
+
+  for (
+    let lowerBound = moment(minDate);
+    lowerBound < cutoffDateObj;
+    lowerBound.add( chunkSize, 'days' )
+  ) {
+    //in case total days period is not devided with chunkSize, the last chunk size needs to be smaller.
+    const remnant = checkRemnant(lowerBound, cutoffDateObj)
+
+    if (remnant < chunkSize && remnant > 0) {
+      chunkSize = remnant;
+    }
+    const upperBound = moment(lowerBound).add( chunkSize, 'days' );
+
+    const lowerBoundId = objectIdFromDate(lowerBound.toDate())
+    const upperBoundId = objectIdFromDate(upperBound.toDate())
+      
+    const logsToArchive = await collectionObj
+      .find({
+        _id: {
+          $gte: ObjectId(lowerBoundId),
+          $lt: ObjectId(upperBoundId),
+        },
+      })
+      .toArray();
+
+    if (logsToArchive.length > 0) {
+      writeLogsToFile(upperBound, lowerBound, collection, logsToArchive, path)
+
+      await collectionObj.deleteMany({
+        _id: {
+          $gte: ObjectId(lowerBoundId),
+          $lt: ObjectId(upperBoundId),
+        },
+      });
+    }
+  }
+}
+
+const command = new Command({
+    command: 'offload-to-file',
+    parent: cmd,
+    description: 'Archiving logs from one or more source collections to files by chunks of days.',
+    webDocs: {
+        category: 'Logs',
+        title: 'Offload To File',
+    },
+    builder: yargs => yargs
+      .option('chunkDuration', {
+        alias: "chdur",
+        describe:
+          "Chunk size in days, each chunk will be archived into a different file.",
+        default: 1,
+        type: "number",
+      })
+      .option('path', {
+        describe: "Directory path to which archive files will be saved.",
+        default: ".",
+        type: "string",
+      })
+      .example('codefresh offline-logs offload-to-file --uri "mongodb://192.168.99.100:27017" --db logs --collections logs foo --cod "2021-07-08" --chdur 3 --path "./" '),
+    handler: async (argv) => {
+      const {
+        uri,
+        cutoffDate,
+        chunkDuration,
+        path,
+        db,
+        collections,
+      } = argv;
+      const client = new MongoClient(uri);
+      try {
+        await client.connect();
+        const failedCollections = [];
+        const database = client.db(db);
+        const promises = collections.map( async ( collection ) => {
+          try {
+            await offloadToFile( database, collection, chunkDuration, cutoffDate, path );
+          } catch (e) {
+            console.log(e)
+            failedCollections.push(collection);
+          }
+        })
+        await Promise.all(promises);
+
+        if (failedCollections.length) {
+          throw new Error(`failed to offload from collections: ${failedCollections.join(', ')}`)
+        }
+      } finally {
+        client.close();
+      }
+    },
+});
+
+module.exports = command;
diff --git a/lib/interface/cli/commands/offline-logs/offload-to-file.spec.js b/lib/interface/cli/commands/offline-logs/offload-to-file.spec.js
@@ -0,0 +1,59 @@
+const moment = require('moment')
+const { writeLogsToFile, checkRemnant } = require('./utils');
+const fs = require('fs')
+
+describe('offload-to-file', () => {
+  describe('writeLogsToFile', () => {
+    it('writes the input data to file', () => {
+      fs.writeFileSync = jest.fn((path, data) => {
+        return {path, data};
+      });
+      const upperBound = '2021-08-10';
+      const lowerBound = '2021-08-01';
+      const collection = 'logs';
+      const logsToArchive = [1, 2, 3, 4];
+      const expectedStr = JSON.stringify(logsToArchive)
+      const path = 'some/path';
+      writeLogsToFile(upperBound, lowerBound, collection, logsToArchive, path);
+      expect(fs.writeFileSync.mock.calls.length).toBe(1);
+      expect(fs.writeFileSync.mock.calls[0][0]).toContain(path);
+      expect(fs.writeFileSync.mock.calls[0][1]).toBe(expectedStr);
+    });
+    it('creates the correct file name', () => {
+      fs.writeFileSync = jest.fn((path, data) => {
+        return {path, data};
+      });
+      const upperBound = '2021-08-10';
+      const lowerBound = '2021-08-01';
+      const collection = 'logs';
+      const logsToArchive = [1, 2, 3, 4];
+      const expectedPath = 'some/path/2021-08-01-2021-08-09-logs.json'
+      const path = 'some/path';
+      writeLogsToFile(upperBound, lowerBound, collection, logsToArchive, path);
+      expect(fs.writeFileSync.mock.calls.length).toBe(1);
+      expect(fs.writeFileSync.mock.calls[0][0]).toBe(expectedPath);
+    });
+    it('checks leap year date calculation', () => {
+      fs.writeFileSync = jest.fn((path, data) => {
+        return {path, data};
+      });
+      const upperBound = '2020-03-01';
+      const lowerBound = '2020-02-01';
+      const collection = 'logs';
+      const logsToArchive = [1, 2, 3, 4];
+      const expectedPath = 'some/path/2020-02-01-2020-02-29-logs.json'
+      const path = 'some/path';
+      writeLogsToFile(upperBound, lowerBound, collection, logsToArchive, path);
+      expect(fs.writeFileSync.mock.calls.length).toBe(1);
+      expect(fs.writeFileSync.mock.calls[0][0]).toBe(expectedPath);
+    });
+  });
+  describe('checkRemnant', () => {
+    it('checkes days remnant to make sure time range is not going off boundaries', () => {
+      const lowerBound = '2020-08-01';
+      const cutoffDateObj = moment('2020-08-01').add(1, 'days').startOf("day");
+      const result = checkRemnant(lowerBound, cutoffDateObj);
+      expect(result).toBe(1);
+    });
+  });
+});
diff --git a/lib/interface/cli/commands/offline-logs/utils.js b/lib/interface/cli/commands/offline-logs/utils.js
@@ -0,0 +1,24 @@
+const moment = require('moment');
+
+const { join } = require('path');
+const fs = require('fs');
+
+// converts date to objectId for filter purposes
+const objectIdFromDate = function (date) {
+	return Math.floor(date.getTime() / 1000).toString(16) + "0000000000000000";
+};
+
+const writeLogsToFile = function(upperBound, lowerBound, collection, logsToArchive, path) {
+  const date = moment(upperBound).subtract(1, "days")
+  const fileDateRange = `${moment(lowerBound).format('YYYY-MM-DD')}-${date.format('YYYY-MM-DD')}`
+  const fileName = `${fileDateRange}-${collection}.json`;
+  const absPath = join(path, fileName);
+  const data = JSON.stringify(logsToArchive);
+  fs.writeFileSync(absPath, data);
+};
+
+const checkRemnant = function(lowerBound, cutoffDateObj) {
+  return Math.ceil(cutoffDateObj.diff(lowerBound , "days" ));
+}
+
+module.exports = {objectIdFromDate, writeLogsToFile, checkRemnant};

Original file line number	Diff line number	Diff line change
`@@ -1,3 +1,5 @@`
	`1`	`+const yargs = require('yargs');`
	`2`	`+`
`1`	`3`	`const Command = require('../../../../Command');`
`2`	`4`	`const createContext = require('../../create.cmd');`
`3`	`5`