rotor tweaks

libs/core-functions/package.json

@@ -40,6 +40,7 @@
     "juava": "workspace:*",
     "lodash": "^4.17.21",
     "mongodb": "^5.9.2",
+    "node-cache": "^5.1.2",
     "@mongodb-js/zstd": "^1.2.0",
     "ioredis": "^5.3.2",
     "posthog-node": "^2.4.0",

libs/core-functions/src/functions/lib/mongodb.ts

@@ -30,7 +30,7 @@ const MongoCreatedCollections = new Set<string>();
 export function mongoAnonymousEventsStore(): AnonymousEventsStore {
   return {
     async addEvent(collectionName: string, anonymousId: string, event: AnalyticsServerEvent, ttlDays: number) {
-      const mongo = await mongodb.waitInit();
+      const mongo = mongodb();
       await ensureMongoCollection(mongo, collectionName, ttlDays);
       const res = await mongo
         .db()
@@ -44,7 +44,7 @@ export function mongoAnonymousEventsStore(): AnonymousEventsStore {
     },
 
     async evictEvents(collectionName: string, anonymousId: string) {
-      const mongo = await mongodb.waitInit();
+      const mongo = mongodb();
       // to ensure query consistency between find and delete query - limit them to the same time window
       const maxObjectId = new ObjectId(Math.floor(new Date().getTime() / 1000).toString(16) + "0000000000000000");
       // load anonymous events from user_recognition collection

libs/core-functions/src/functions/lib/ua.ts

@@ -1,16 +1,30 @@
 import { UserAgent } from "@jitsu/protocols/functions";
 import omit from "lodash/omit";
 import uaParser from "@amplitude/ua-parser-js";
+import NodeCache from "node-cache";
 
 const BotUAKeywords = ["bot", "spider", "headless", "crawler", "uptimia"];
+const uaCacheTTL = 60 * 10; // 10 min;
+const uaCache = new NodeCache({ stdTTL: uaCacheTTL, checkperiod: 60, maxKeys: 10000 });
 
 export function parseUserAgent(userAgent?: string): UserAgent {
+  if (!userAgent) {
+    return {} as UserAgent;
+  }
+  const cached = uaCache.get(userAgent);
+  if (cached) {
+    uaCache.ttl(userAgent, uaCacheTTL);
+    return cached as UserAgent;
+  }
   const uas = userAgent || "";
   const ua = omit(uaParser(uas), "ua") as UserAgent;
   const lower = uas.toLowerCase();
   ua.bot = BotUAKeywords.some(keyword => lower.includes(keyword));
   if (ua.device) {
     ua.device.type = ua.device.type || "desktop";
   }
+  try {
+    uaCache.set(userAgent, ua);
+  } catch (e) {}
   return ua;
 }

services/rotor/src/index.ts

@@ -17,6 +17,7 @@ import { FunctionsHandler, FunctionsHandlerMulti } from "./http/functions";
 import { initMaxMindClient } from "./lib/maxmind";
 import { rotorMessageHandler } from "./lib/message-handler";
 import { redis } from "@jitsu-internal/console/lib/server/redis";
+import { redisLogger } from "./lib/redis-logger";
 
 export const log = getLog("rotor");
 
@@ -57,6 +58,7 @@ async function main() {
     Prometheus.collectDefaultMetrics();
     await mongodb.waitInit();
     await redis.waitInit();
+    await redisLogger.waitInit();
     rotor
       .start()
       .then(chMetrics => {

services/rotor/src/lib/message-handler.ts

@@ -115,7 +115,7 @@ export async function rotorMessageHandler(
     ? createMultiStore(newStore, oldStore)
     : oldStore;
 
-  const rl = await redisLogger.waitInit();
+  const rl = redisLogger();
 
   const connectionData = connection.options as any;