Merge pull request #1471 from akhilmhdh/feat/telemetry-aggregation

Telemetry stats event for self hosted instance on midnight

backend/src/services/super-admin/super-admin-service.ts

@@ -1,4 +1,5 @@
 import { TSuperAdmin, TSuperAdminUpdate } from "@app/db/schemas";
+import { TKeyStoreFactory } from "@app/keystore/keystore";
 import { getConfig } from "@app/lib/config/env";
 import { BadRequestError } from "@app/lib/errors";
 
@@ -14,6 +15,7 @@ type TSuperAdminServiceFactoryDep = {
   userDAL: TUserDALFactory;
   authService: Pick<TAuthLoginFactory, "generateUserTokens">;
   orgService: Pick<TOrgServiceFactory, "createOrganization">;
+  keyStore: Pick<TKeyStoreFactory, "getItem" | "setItemWithExpiry" | "deleteItem">;
 };
 
 export type TSuperAdminServiceFactory = ReturnType<typeof superAdminServiceFactory>;
@@ -21,26 +23,53 @@ export type TSuperAdminServiceFactory = ReturnType<typeof superAdminServiceFacto
 // eslint-disable-next-line
 export let getServerCfg: () => Promise<TSuperAdmin>;
 
+const ADMIN_CONFIG_KEY = "infisical-admin-cfg";
+const ADMIN_CONFIG_KEY_EXP = 60; // 60s
+const ADMIN_CONFIG_DB_UUID = "00000000-0000-0000-0000-000000000000";
+
 export const superAdminServiceFactory = ({
   serverCfgDAL,
   userDAL,
   authService,
-  orgService
+  orgService,
+  keyStore
 }: TSuperAdminServiceFactoryDep) => {
   const initServerCfg = async () => {
     // TODO(akhilmhdh): bad  pattern time less change this later to me itself
-    getServerCfg = () => serverCfgDAL.findOne({});
+    getServerCfg = async () => {
+      const config = await keyStore.getItem(ADMIN_CONFIG_KEY);
+      // missing in keystore means fetch from db
+      if (!config) {
+        const serverCfg = await serverCfgDAL.findById(ADMIN_CONFIG_DB_UUID);
+        if (serverCfg) {
+          await keyStore.setItemWithExpiry(ADMIN_CONFIG_KEY, ADMIN_CONFIG_KEY_EXP, JSON.stringify(serverCfg)); // insert it back to keystore
+        }
+        return serverCfg;
+      }
 
-    const serverCfg = await serverCfgDAL.findOne({});
+      const keyStoreServerCfg = JSON.parse(config) as TSuperAdmin;
+      return {
+        ...keyStoreServerCfg,
+        // this is to allow admin router to work
+        createdAt: new Date(keyStoreServerCfg.createdAt),
+        updatedAt: new Date(keyStoreServerCfg.updatedAt)
+      };
+    };
+
+    // reset on initialized
+    await keyStore.deleteItem(ADMIN_CONFIG_KEY);
+    const serverCfg = await serverCfgDAL.findById(ADMIN_CONFIG_DB_UUID);
     if (serverCfg) return;
-    const newCfg = await serverCfgDAL.create({ initialized: false, allowSignUp: true });
+
+    // @ts-expect-error id is kept as fixed for idempotence and to avoid race condition
+    const newCfg = await serverCfgDAL.create({ initialized: false, allowSignUp: true, id: ADMIN_CONFIG_DB_UUID });
     return newCfg;
   };
 
   const updateServerCfg = async (data: TSuperAdminUpdate) => {
-    const serverCfg = await getServerCfg();
-    const cfg = await serverCfgDAL.updateById(serverCfg.id, data);
-    return cfg;
+    const updatedServerCfg = await serverCfgDAL.updateById(ADMIN_CONFIG_DB_UUID, data);
+    await keyStore.setItemWithExpiry(ADMIN_CONFIG_KEY, ADMIN_CONFIG_KEY_EXP, JSON.stringify(updatedServerCfg));
+    return updatedServerCfg;
   };
 
   const adminSignUp = async ({

backend/src/services/telemetry/telemetry-dal.ts

@@ -0,0 +1,39 @@
+import { TDbClient } from "@app/db";
+import { TableName } from "@app/db/schemas";
+import { DatabaseError } from "@app/lib/errors";
+
+export type TTelemetryDALFactory = ReturnType<typeof telemetryDALFactory>;
+
+export const telemetryDALFactory = (db: TDbClient) => {
+  const getTelemetryInstanceStats = async () => {
+    try {
+      const userCount = (await db(TableName.Users).where({ isGhost: false }).count().first())?.count as string;
+      const users = parseInt(userCount || "0", 10);
+
+      const identityCount = (await db(TableName.Identity).count().first())?.count as string;
+      const identities = parseInt(identityCount || "0", 10);
+
+      const projectCount = (await db(TableName.Project).count().first())?.count as string;
+      const projects = parseInt(projectCount || "0", 10);
+
+      const secretCount = (await db(TableName.Secret).count().first())?.count as string;
+      const secrets = parseInt(secretCount || "0", 10);
+
+      const organizationNames = await db(TableName.Organization).select("name");
+      const organizations = organizationNames.length;
+
+      return {
+        users,
+        identities,
+        projects,
+        secrets,
+        organizations,
+        organizationNames: organizationNames.map(({ name }) => name)
+      };
+    } catch (error) {
+      throw new DatabaseError({ error, name: "TelemtryInstanceStats" });
+    }
+  };
+
+  return { getTelemetryInstanceStats };
+};

backend/src/services/telemetry/telemetry-queue.ts

@@ -0,0 +1,78 @@
+import { PostHog } from "posthog-node";
+
+import { TKeyStoreFactory } from "@app/keystore/keystore";
+import { getConfig } from "@app/lib/config/env";
+import { logger } from "@app/lib/logger";
+import { QueueJobs, QueueName, TQueueServiceFactory } from "@app/queue";
+
+import { getServerCfg } from "../super-admin/super-admin-service";
+import { TTelemetryDALFactory } from "./telemetry-dal";
+import { TELEMETRY_SECRET_OPERATIONS_KEY, TELEMETRY_SECRET_PROCESSED_KEY } from "./telemetry-service";
+import { PostHogEventTypes } from "./telemetry-types";
+
+type TTelemetryQueueServiceFactoryDep = {
+  queueService: TQueueServiceFactory;
+  keyStore: Pick<TKeyStoreFactory, "getItem" | "deleteItem">;
+  telemetryDAL: TTelemetryDALFactory;
+};
+
+export type TTelemetryQueueServiceFactory = ReturnType<typeof telemetryQueueServiceFactory>;
+
+export const telemetryQueueServiceFactory = ({
+  queueService,
+  keyStore,
+  telemetryDAL
+}: TTelemetryQueueServiceFactoryDep) => {
+  const appCfg = getConfig();
+  const postHog =
+    appCfg.isProductionMode && appCfg.TELEMETRY_ENABLED
+      ? new PostHog(appCfg.POSTHOG_PROJECT_API_KEY, { host: appCfg.POSTHOG_HOST, flushAt: 1, flushInterval: 0 })
+      : undefined;
+
+  queueService.start(QueueName.TelemetryInstanceStats, async () => {
+    const { instanceId } = await getServerCfg();
+    const telemtryStats = await telemetryDAL.getTelemetryInstanceStats();
+    // parse the redis values into integer
+    const numberOfSecretOperationsMade = parseInt((await keyStore.getItem(TELEMETRY_SECRET_OPERATIONS_KEY)) || "0", 10);
+    const numberOfSecretProcessed = parseInt((await keyStore.getItem(TELEMETRY_SECRET_PROCESSED_KEY)) || "0", 10);
+    const stats = { ...telemtryStats, numberOfSecretProcessed, numberOfSecretOperationsMade };
+
+    // send to postHog
+    postHog?.capture({
+      event: PostHogEventTypes.TelemetryInstanceStats,
+      distinctId: instanceId,
+      properties: stats
+    });
+    // reset the stats
+    await keyStore.deleteItem(TELEMETRY_SECRET_PROCESSED_KEY);
+    await keyStore.deleteItem(TELEMETRY_SECRET_OPERATIONS_KEY);
+  });
+
+  // every day at midnight a telemetry job executes on self hosted
+  // this sends some telemetry information like instance id secrets operated etc
+  const startTelemetryCheck = async () => {
+    // this is a fast way to check its cloud or not
+    if (appCfg.INFISICAL_CLOUD) return;
+    // clear previous job
+    await queueService.stopRepeatableJob(
+      QueueName.TelemetryInstanceStats,
+      QueueJobs.TelemetryInstanceStats,
+      { pattern: "0 0 * * *", utc: true },
+      QueueName.TelemetryInstanceStats // just a job id
+    );
+    if (postHog) {
+      await queueService.queue(QueueName.TelemetryInstanceStats, QueueJobs.TelemetryInstanceStats, undefined, {
+        jobId: QueueName.TelemetryInstanceStats,
+        repeat: { pattern: "0 0 * * *", utc: true }
+      });
+    }
+  };
+
+  queueService.listen(QueueName.TelemetryInstanceStats, "failed", (err) => {
+    logger.error(err?.failedReason, `${QueueName.TelemetryInstanceStats}: failed`);
+  });
+
+  return {
+    startTelemetryCheck
+  };
+};

backend/src/services/telemetry/telemetry-service.ts

@@ -1,15 +1,24 @@
 import { PostHog } from "posthog-node";
 
+import { TLicenseServiceFactory } from "@app/ee/services/license/license-service";
+import { InstanceType } from "@app/ee/services/license/license-types";
+import { TKeyStoreFactory } from "@app/keystore/keystore";
 import { getConfig } from "@app/lib/config/env";
 import { request } from "@app/lib/config/request";
 import { logger } from "@app/lib/logger";
 
-import { TPostHogEvent } from "./telemetry-types";
+import { PostHogEventTypes, TPostHogEvent, TSecretModifiedEvent } from "./telemetry-types";
+
+export const TELEMETRY_SECRET_PROCESSED_KEY = "telemetry-secret-processed";
+export const TELEMETRY_SECRET_OPERATIONS_KEY = "telemetry-secret-operations";
 
 export type TTelemetryServiceFactory = ReturnType<typeof telemetryServiceFactory>;
+export type TTelemetryServiceFactoryDep = {
+  keyStore: Pick<TKeyStoreFactory, "getItem" | "incrementBy">;
+  licenseService: Pick<TLicenseServiceFactory, "getInstanceType">;
+};
 
-// type TTelemetryServiceFactoryDep = {};
-export const telemetryServiceFactory = () => {
+export const telemetryServiceFactory = ({ keyStore, licenseService }: TTelemetryServiceFactoryDep) => {
   const appCfg = getConfig();
 
   if (appCfg.isProductionMode && !appCfg.TELEMETRY_ENABLED) {
@@ -21,10 +30,9 @@ To opt into telemetry, you can set "TELEMETRY_ENABLED=true" within the environme
 `);
   }
 
-  const postHog =
-    appCfg.isProductionMode && appCfg.TELEMETRY_ENABLED
-      ? new PostHog(appCfg.POSTHOG_PROJECT_API_KEY, { host: appCfg.POSTHOG_HOST })
-      : undefined;
+  const postHog = appCfg.TELEMETRY_ENABLED
+    ? new PostHog(appCfg.POSTHOG_PROJECT_API_KEY, { host: appCfg.POSTHOG_HOST })
+    : undefined;
 
   // used for email marketting email sending purpose
   const sendLoopsEvent = async (email: string, firstName?: string, lastName?: string) => {
@@ -51,13 +59,33 @@ To opt into telemetry, you can set "TELEMETRY_ENABLED=true" within the environme
     }
   };
 
-  const sendPostHogEvents = (event: TPostHogEvent) => {
+  const sendPostHogEvents = async (event: TPostHogEvent) => {
     if (postHog) {
-      postHog.capture({
-        event: event.event,
-        distinctId: event.distinctId,
-        properties: event.properties
-      });
+      const instanceType = licenseService.getInstanceType();
+      // capture posthog only when its cloud or signup event happens in self hosted
+      if (instanceType === InstanceType.Cloud || event.event === PostHogEventTypes.UserSignedUp) {
+        postHog.capture({
+          event: event.event,
+          distinctId: event.distinctId,
+          properties: event.properties
+        });
+        return;
+      }
+
+      if (
+        [
+          PostHogEventTypes.SecretPulled,
+          PostHogEventTypes.SecretCreated,
+          PostHogEventTypes.SecretDeleted,
+          PostHogEventTypes.SecretUpdated
+        ].includes(event.event)
+      ) {
+        await keyStore.incrementBy(
+          TELEMETRY_SECRET_PROCESSED_KEY,
+          (event as TSecretModifiedEvent).properties.numberOfSecrets
+        );
+        await keyStore.incrementBy(TELEMETRY_SECRET_OPERATIONS_KEY, 1);
+      }
     }
   };
 

backend/src/services/telemetry/telemetry-types.ts

@@ -12,7 +12,8 @@ export enum PostHogEventTypes {
   ProjectCreated = "Project Created",
   IntegrationCreated = "Integration Created",
   MachineIdentityCreated = "Machine Identity Created",
-  UserOrgInvitation = "User Org Invitation"
+  UserOrgInvitation = "User Org Invitation",
+  TelemetryInstanceStats = "Self Hosted Instance Stats"
 }
 
 export type TSecretModifiedEvent = {
@@ -101,6 +102,20 @@ export type TUserOrgInvitedEvent = {
   };
 };
 
+export type TTelemetryInstanceStatsEvent = {
+  event: PostHogEventTypes.TelemetryInstanceStats;
+  properties: {
+    users: number;
+    identities: number;
+    projects: number;
+    secrets: number;
+    organizations: number;
+    organizationNames: number;
+    numberOfSecretOperationsMade: number;
+    numberOfSecretProcessed: number;
+  };
+};
+
 export type TPostHogEvent = { distinctId: string } & (
   | TSecretModifiedEvent
   | TAdminInitEvent
@@ -110,4 +125,5 @@ export type TPostHogEvent = { distinctId: string } & (
   | TMachineIdentityCreatedEvent
   | TIntegrationCreatedEvent
   | TProjectCreateEvent
+  | TTelemetryInstanceStatsEvent
 );