Add wrapper around container logs saving, to reduce db load

Changes are collected together and exist in memory, for querying and
saving. Once every 10 mins, every changed timestamp is flushed to the
database.

Change-type: patch
Closes: #987
Signed-off-by: Cameron Diver <cameron@balena.io>

src/logger.ts

@@ -13,6 +13,7 @@ import {
 	LogBackend,
 	LogMessage,
 } from './logging';
+import LogMonitor from './logging/monitor';
 
 interface LoggerSetupOptions {
 	apiEndpoint: string;
@@ -38,11 +39,13 @@ export class Logger {
 	private eventTracker: EventTracker;
 	private db: DB;
 	private containerLogs: { [containerId: string]: ContainerLogs } = {};
+	private logMonitor: LogMonitor;
 
 	public constructor({ db, eventTracker }: LoggerConstructOptions) {
 		this.backend = null;
 		this.eventTracker = eventTracker;
 		this.db = db;
+		this.logMonitor = new LogMonitor(db);
 	}
 
 	public init({
@@ -151,28 +154,17 @@ export class Logger {
 
 				// Take the timestamp and set it in the database as the last
 				// log sent for this
-				await this.db
-					.models('containerLogs')
-					.where({ containerId })
-					.update({ lastSentTimestamp: logMessage.timestamp });
+				this.logMonitor.updateContainerSentTimestamp(
+					containerId,
+					logMessage.timestamp,
+				);
 			});
 
 			logs.on('closed', () => delete this.containerLogs[containerId]);
 
-			// Get the timestamp of the last sent log for this container
-			let [timestampObj] = await this.db
-				.models('containerLogs')
-				.select('lastSentTimestamp')
-				.where({ containerId });
-
-			if (timestampObj == null) {
-				timestampObj = { lastSentTimestamp: 0 };
-				// Create the row so we have something to update
-				await this.db
-					.models('containerLogs')
-					.insert({ containerId, lastSentTimestamp: 0 });
-			}
-			const { lastSentTimestamp } = timestampObj;
+			const lastSentTimestamp = await this.logMonitor.getContainerSentTimestamp(
+				containerId,
+			);
 			return logs.attach(lastSentTimestamp);
 		});
 	}

src/logging/monitor.ts

@@ -0,0 +1,82 @@
+import Database from '../db';
+
+// Flush every 10 mins
+const DB_FLUSH_INTERVAL = 10 * 60 * 1000;
+
+/**
+ * This class provides a wrapper around the database for
+ * saving the last timestamp of a container
+ */
+export class LogMonitor {
+	private timestamps: { [containerId: string]: number } = {};
+	private writeRequired: { [containerId: string]: boolean } = {};
+
+	public constructor(private db: Database) {
+		setInterval(() => this.flushDb(), DB_FLUSH_INTERVAL);
+	}
+
+	public updateContainerSentTimestamp(
+		containerId: string,
+		timestamp: number,
+	): void {
+		this.timestamps[containerId] = timestamp;
+		this.writeRequired[containerId] = true;
+	}
+
+	public async getContainerSentTimestamp(containerId: string): Promise<number> {
+		// If this is the first time we are requesting the
+		// timestamp for this container, request it from the db
+		if (this.timestamps[containerId] == null) {
+			// Set the timestamp to 0 before interacting with the
+			// db to avoid multiple db actions at once
+			this.timestamps[containerId] = 0;
+			try {
+				const timestampObj = await this.db
+					.models('containerLogs')
+					.select('lastSentTimestamp')
+					.where({ containerId });
+
+				if (timestampObj == null) {
+					// Create a row in the db so there's something to
+					// update
+					await this.db
+						.models('containerLogs')
+						.insert({ containerId, lastSentTimestamp: 0 });
+				} else {
+					this.timestamps[containerId] = timestampObj.lastSentTimestamp;
+				}
+			} catch (e) {
+				console.error(
+					`There was an error retrieving the container log timestamps: ${e}`,
+				);
+			}
+		}
+		return this.timestamps[containerId] || 0;
+	}
+
+	private async flushDb() {
+		console.log('Attempting container log timestamp flush...');
+		const containerIds = Object.getOwnPropertyNames(this.timestamps);
+		try {
+			for (const containerId of containerIds) {
+				// Avoid writing to the db if we don't need to
+				if (!this.writeRequired[containerId]) {
+					continue;
+				}
+
+				await this.db
+					.models('containerLogs')
+					.where({ containerId })
+					.update({ lastSentTimestamp: this.timestamps[containerId] });
+				this.writeRequired[containerId] = false;
+			}
+		} catch (e) {
+			console.error(
+				`There was an error storing the container log timestamps: ${e}`,
+			);
+		}
+		console.log('Container log timestamp flush complete');
+	}
+}
+
+export default LogMonitor;