Merge pull request #988 from balena-io/987-knex-timeout

Add in-memory cache around container logs saving, to reduce db load

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
+				this.logMonitor.updateContainerSentTimestamp(
-					.models('containerLogs')
+					containerId,
-					.where({ containerId })
+					logMessage.timestamp,
-					.update({ lastSentTimestamp: logMessage.timestamp });
+				);
 			});
 
 			logs.on('closed', () => delete this.containerLogs[containerId]);
 
-			// Get the timestamp of the last sent log for this container
+			const lastSentTimestamp = await this.logMonitor.getContainerSentTimestamp(
-			let [timestampObj] = await this.db
+				containerId,
-				.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;
 			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;