rotor: get rid of pg-config-store in favor of http based entity-store

absorbb · absorbb · commit 5b8a42966117 · 2024-01-19T16:40:00.000+04:00
diff --git a/services/rotor/src/index.ts b/services/rotor/src/index.ts
@@ -20,7 +20,7 @@ import { redis } from "./lib/redis";
 import { redisLogger } from "./lib/redis-logger";
 import { DummyMetrics, Metrics } from "./lib/metrics";
 import { isTruish } from "@jitsu-internal/console/lib/shared/chores";
-import { pgConfigStore } from "./lib/pg-config-store";
+import { connectionsStore, functionsStore } from "./lib/entity-store";
 
 export const log = getLog("rotor");
 
@@ -67,9 +67,14 @@ async function main() {
     await mongodb.waitInit();
     await redis.waitInit();
     await redisLogger.waitInit();
-    const store = await pgConfigStore.get();
-    if (!store.enabled) {
-      log.atError().log("Postgres is not configured. Rotor will not work");
+    const connStore = await connectionsStore.get();
+    if (!connStore.enabled) {
+      log.atError().log("Connection store is not configured. Rotor will not work");
+      process.exit(1);
+    }
+    const funcStore = await functionsStore.get();
+    if (!funcStore.enabled) {
+      log.atError().log("Functions store is not configured. Rotor will not work");
       process.exit(1);
     }
     //kafka consumer mode
@@ -104,9 +109,14 @@ async function main() {
     await mongodb.waitInit();
     await redis.waitInit();
     await redisLogger.waitInit();
-    const store = await pgConfigStore.get();
-    if (!store.enabled) {
-      log.atError().log("Postgres is not configured. Rotor will not work");
+    const connStore = await connectionsStore.get();
+    if (!connStore.enabled) {
+      log.atError().log("Connection store is not configured. Rotor will not work");
+      process.exit(1);
+    }
+    const funcStore = await functionsStore.get();
+    if (!funcStore.enabled) {
+      log.atError().log("Functions store is not configured. Rotor will not work");
       process.exit(1);
     }
     const geoResolver = await initMaxMindClient(process.env.MAXMIND_LICENSE_KEY || "");
@@ -118,11 +128,17 @@ function initHTTP(metrics: Metrics, geoResolver: GeoResolver) {
   http.get("/health", (req, res) => {
     res.json({
       status: "pass",
-      configStore: {
-        enabled: pgConfigStore.getCurrent()?.enabled || "loading",
-        status: pgConfigStore.status(),
-        lastUpdated: pgConfigStore.lastRefresh(),
-        lastModified: pgConfigStore.lastModified(),
+      connectionsStore: {
+        enabled: connectionsStore.getCurrent()?.enabled || "loading",
+        status: connectionsStore.status(),
+        lastUpdated: connectionsStore.lastRefresh(),
+        lastModified: connectionsStore.lastModified(),
+      },
+      functionsStore: {
+        enabled: functionsStore.getCurrent()?.enabled || "loading",
+        status: functionsStore.status(),
+        lastUpdated: functionsStore.lastRefresh(),
+        lastModified: functionsStore.lastModified(),
       },
     });
   });
diff --git a/services/rotor/src/lib/entity-store.ts b/services/rotor/src/lib/entity-store.ts
@@ -0,0 +1,113 @@
+import { createInMemoryStore } from "./inmem-store";
+import { getLog } from "juava";
+import { httpAgent, httpsAgent } from "@jitsu/core-functions";
+import fetch from "node-fetch-commonjs";
+
+const log = getLog("entity-store");
+
+export type EntityStore = {
+  getObject: (id: string) => any;
+  toJSON: () => string;
+  enabled: boolean;
+};
+
+const DisabledStore: EntityStore = {
+  enabled: false,
+  getObject: () => undefined,
+  toJSON: () => "disabled",
+};
+
+const EmptyStore: EntityStore = {
+  enabled: true,
+  getObject: () => undefined,
+  toJSON: () => "",
+};
+
+const refreshFunc =
+  (storeId: string) =>
+  async (ifModifiedSince?: Date): Promise<{ lastModified: Date | undefined; store: EntityStore } | "not_modified"> => {
+    const repositoryBase = process.env.REPOSITORY_BASE_URL;
+    if (repositoryBase) {
+      const objs: Record<string, any> = {};
+      const headers: Record<string, string> = {};
+      let lastModified: Date | undefined = undefined;
+      if (process.env.REPOSITORY_AUTH_TOKEN) {
+        headers["Authorization"] = `Bearer ${process.env.REPOSITORY_AUTH_TOKEN}`;
+      }
+      if (ifModifiedSince) {
+        headers["If-Modified-Since"] = ifModifiedSince.toUTCString();
+      }
+      try {
+        const res = await fetch(`${repositoryBase}/${storeId}?timeoutMs=10000&listen=1`, {
+          method: "GET",
+          headers: headers,
+          agent: await (repositoryBase.startsWith("https://") ? httpsAgent : httpAgent).waitInit(),
+        });
+        if (res.status === 304) {
+          log.atInfo().log(`${storeId} nod modified: ${ifModifiedSince}`);
+          return "not_modified";
+        }
+        if (!res.ok) {
+          throw new Error(`Failed to load ${storeId} from repository: ${res.status} response: ${await res.text()}`);
+        }
+        const json: any = await res.json();
+        for (const fn of json) {
+          objs[fn.id] = fn;
+        }
+        const lmString = res.headers.get("Last-Modified");
+        if (lmString) {
+          lastModified = new Date(lmString);
+        }
+        log.atInfo().log(`${storeId} updated: ${lastModified} previous update date: ${ifModifiedSince}`);
+        return {
+          store: {
+            enabled: true,
+            getObject: (key: string) => {
+              return objs[key];
+            },
+            toJSON: () => {
+              return JSON.stringify(objs);
+            },
+          },
+          lastModified: lastModified,
+        };
+      } catch (e) {
+        throw new Error(`Failed to load ${storeId} from repository: ${e}`);
+      }
+    } else {
+      return { store: DisabledStore, lastModified: new Date() };
+    }
+  };
+
+const storeFunc = (storeId: string) =>
+  createInMemoryStore({
+    refreshIntervalMillis: process.env.REPOSITORY_REFRESH_PERIOD_SEC
+      ? parseInt(process.env.REPOSITORY_REFRESH_PERIOD_SEC) * 1000
+      : 2000,
+    name: `${storeId}-store`,
+    localDir: process.env.REPOSITORY_CACHE_DIR,
+    serializer: (store: EntityStore) => (store.enabled ? store.toJSON() : ""),
+    deserializer: (serialized: string) => {
+      if (serialized) {
+        if (serialized === "disabled") {
+          return DisabledStore;
+        }
+        const store = JSON.parse(serialized);
+        return {
+          enabled: true,
+          getObject: (key: string): any => {
+            return store?.[key];
+          },
+          toJSON: () => {
+            return store ? JSON.stringify(store) : "";
+          },
+        };
+      } else {
+        return EmptyStore;
+      }
+    },
+    refresh: refreshFunc(storeId),
+  });
+
+export const functionsStore = storeFunc("functions");
+export const connectionsStore = storeFunc("rotor-connections");
diff --git a/services/rotor/src/lib/functions-chain.ts b/services/rotor/src/lib/functions-chain.ts
@@ -19,10 +19,10 @@ import { RetryErrorName, DropRetryErrorName } from "@jitsu/functions-lib";
 
 import { getLog, newError, requireDefined, stopwatch } from "juava";
 import { retryObject } from "./retries";
-import { ConfigStore } from "./pg-config-store";
 import NodeCache from "node-cache";
 import pick from "lodash/pick";
 import { EnrichedConnectionConfig } from "./config-types";
+import { EntityStore } from "./entity-store";
 
 export type Func = {
   id: string;
@@ -87,7 +87,7 @@ const getCachedOrLoad = (cache: NodeCache, key: string, loader: (key: string) =>
 
 export function buildFunctionChain(
   connection: EnrichedConnectionConfig,
-  pgStore: ConfigStore,
+  func: EntityStore,
   functionsFilter?: (id: string) => boolean
 ) {
   let mainFunction;
@@ -123,7 +123,7 @@ export function buildFunctionChain(
     .filter(f => f.functionId.startsWith("udf."))
     .map(f => {
       const functionId = f.functionId.substring(4);
-      const userFunctionObj = pgStore.getConfig("function", functionId);
+      const userFunctionObj = func.getObject(functionId);
       if (!userFunctionObj || userFunctionObj.workspaceId !== connection.workspaceId) {
         return {
           id: f.functionId as string,
diff --git a/services/rotor/src/lib/inmem-store.ts b/services/rotor/src/lib/inmem-store.ts
@@ -62,10 +62,7 @@ export const createInMemoryStore = <T>(definition: StoreDefinition<T>): InMemory
   let intervalToClear: NodeJS.Timeout | undefined = undefined;
 
   function scheduleStoreRefresh() {
-    intervalToClear = setInterval(async () => {
-      if (stopping) {
-        return;
-      }
+    const refresh = async () => {
       try {
         const newDef = await definition.refresh(lastModified);
         if (newDef !== "not_modified") {
@@ -79,7 +76,25 @@ export const createInMemoryStore = <T>(definition: StoreDefinition<T>): InMemory
         log.atWarn().withCause(e).log(`Failed to refresh store ${definition.name}. Using an old value`);
         status = "outdated";
       }
-    }, definition.refreshIntervalMillis);
+    };
+    if (definition.refreshIntervalMillis > 0) {
+      intervalToClear = setInterval(async () => {
+        if (stopping) {
+          return;
+        }
+        await refresh();
+      }, definition.refreshIntervalMillis);
+    } else {
+      (function loop() {
+        if (stopping) {
+          return;
+        }
+        setTimeout(async () => {
+          await refresh();
+          loop();
+        }, 1);
+      })();
+    }
   }
 
   let instancePromise = new Promise<T>((resolve, reject) => {
@@ -149,8 +164,6 @@ export const createInMemoryStore = <T>(definition: StoreDefinition<T>): InMemory
       stopping = true;
       if (intervalToClear) {
         clearInterval(intervalToClear);
-      } else {
-        log.atError().log(`There's no interval for ${definition.name}`);
       }
       status = "stopped";
     },
diff --git a/services/rotor/src/lib/message-handler.ts b/services/rotor/src/lib/message-handler.ts
@@ -3,7 +3,8 @@ import { Metrics } from "./metrics";
 import { GeoResolver } from "./maxmind";
 import { IngestMessage } from "@jitsu/protocols/async-request";
 import { CONNECTION_IDS_HEADER } from "./rotor";
-import { pgConfigStore } from "./pg-config-store";
+import { connectionsStore, functionsStore } from "./entity-store";
+
 import { AnalyticsServerEvent } from "@jitsu/protocols/analytics";
 import { EventContext } from "@jitsu/protocols/functions";
 import {
@@ -36,17 +37,21 @@ export async function rotorMessageHandler(
   if (!_message) {
     return;
   }
-  const pgStore = pgConfigStore.getCurrent();
-  if (!pgStore || !pgStore.enabled) {
-    throw newError("Config store is not enabled");
+  const connStore = connectionsStore.getCurrent();
+  if (!connStore || !connStore.enabled) {
+    throw newError("Connection store is not enabled");
+  }
+  const funcStore = functionsStore.getCurrent();
+  if (!funcStore || !funcStore.enabled) {
+    throw newError("Functions store is not enabled");
   }
   const eventStore = redisLogger();
 
   const message = (typeof _message === "string" ? JSON.parse(_message) : _message) as IngestMessage;
   const connectionId =
     headers && headers[CONNECTION_IDS_HEADER] ? headers[CONNECTION_IDS_HEADER].toString() : message.connectionId;
   const connection: EnrichedConnectionConfig = requireDefined(
-    pgStore.getEnrichedConnection(connectionId),
+    connStore.getObject(connectionId),
     `Unknown connection: ${connectionId}`
   );
 
@@ -107,7 +112,7 @@ export async function rotorMessageHandler(
     },
   };
 
-  const funcChain = buildFunctionChain(connection, pgStore, functionsFilter);
+  const funcChain = buildFunctionChain(connection, funcStore, functionsFilter);
 
   const chainRes = await runChain(funcChain, event, eventStore, store, ctx, systemContext);
   chainRes.connectionId = connectionId;
diff --git a/services/rotor/src/lib/pg-config-store.ts b/services/rotor/src/lib/pg-config-store.ts
diff --git a/services/rotor/src/lib/rotor.ts b/services/rotor/src/lib/rotor.ts
