Event property counter

latest_migrations.manifest

@@ -3,7 +3,7 @@ auth: 0012_alter_user_first_name_max_length
 axes: 0006_remove_accesslog_trusted
 contenttypes: 0002_remove_content_type_name
 ee: 0005_project_based_permissioning
-posthog: 0191_rename_specialmigration_asyncmigration
+posthog: 0192_event_properties
 rest_hooks: 0002_swappable_hook_model
 sessions: 0001_initial
 social_django: 0010_uid_db_index

plugin-server/package.json

@@ -33,7 +33,7 @@
         "prepublishOnly": "yarn build",
         "setup:dev:clickhouse": "cd .. && export DEBUG=1 PRIMARY_DB=clickhouse && python manage.py migrate_clickhouse",
         "setup:test:ee": "yarn setup:test:postgres && yarn setup:test:clickhouse",
-        "setup:test:postgres": "cd .. && (dropdb test_posthog || echo 'no db to drop') && createdb test_posthog && DATABASE_URL=postgres://localhost:5432/test_posthog DEBUG=1 python manage.py migrate",
+        "setup:test:postgres": "cd .. && (PGPASSWORD=posthog dropdb -h localhost -U posthog test_posthog || echo 'no db to drop') && PGPASSWORD=posthog createdb -h localhost -U posthog test_posthog && DATABASE_URL=postgres://posthog:posthog@localhost:5432/test_posthog DEBUG=1 python manage.py migrate",
         "setup:test:clickhouse": "cd .. && unset KAFKA_URL && export TEST=1 PRIMARY_DB=clickhouse CLICKHOUSE_DATABASE=posthog_test && python manage.py migrate_clickhouse",
         "services:start": "cd .. && docker-compose -f ee/docker-compose.ch.yml up zookeeper kafka clickhouse",
         "services:stop": "cd .. && docker-compose -f ee/docker-compose.ch.yml down",

plugin-server/src/config/config.ts

@@ -15,9 +15,9 @@ export function getDefaultConfig(): PluginsServerConfig {
     return {
         CELERY_DEFAULT_QUEUE: 'celery',
         DATABASE_URL: isTestEnv
-            ? 'postgres://localhost:5432/test_posthog'
+            ? 'postgres://posthog:posthog@localhost:5432/test_posthog'
             : isDevEnv
-            ? 'postgres://localhost:5432/posthog'
+            ? 'postgres://posthog:posthog@localhost:5432/posthog'
             : null,
         POSTHOG_DB_NAME: null,
         POSTHOG_DB_USER: 'postgres',
@@ -78,6 +78,7 @@ export function getDefaultConfig(): PluginsServerConfig {
         SITE_URL: null,
         NEW_PERSON_PROPERTIES_UPDATE_ENABLED_TEAMS: '',
         EXPERIMENTAL_EVENTS_LAST_SEEN_ENABLED: true,
+        EXPERIMENTAL_EVENT_PROPERTY_COUNTER_ENABLED_TEAMS: '',
     }
 }
 
@@ -135,7 +136,9 @@ export function getConfigHelp(): Record<keyof PluginsServerConfig, string> {
             '(advanced) corresponds to the length of time a piscina worker should block for when looking for tasks',
         NEW_PERSON_PROPERTIES_UPDATE_ENABLED_TEAMS:
             '(advanced) teams for which to run the new person properties update flow on',
-        EXPERIMENTAL_EVENTS_LAST_SEEN_ENABLED: 'enable experimental feature to track lastSeenAt',
+        EXPERIMENTAL_EVENTS_LAST_SEEN_ENABLED: '(advanced) enable experimental feature to track lastSeenAt',
+        EXPERIMENTAL_EVENT_PROPERTY_COUNTER_ENABLED_TEAMS:
+            '(advanced) teams for which to enable experimental feature to count event properties',
     }
 }
 

plugin-server/src/main/pluginsServer.ts

@@ -47,6 +47,7 @@ export async function startPluginsServer(
     let piscinaStatsJob: schedule.Job | undefined
     let internalMetricsStatsJob: schedule.Job | undefined
     let flushLastSeenAtCacheJob: schedule.Job | undefined
+    let flushEventPropertyCounterJob: schedule.Job | undefined
     let pluginMetricsJob: schedule.Job | undefined
     let piscina: Piscina | undefined
     let queue: Queue | undefined // ingestion queue
@@ -81,6 +82,8 @@ export async function startPluginsServer(
         statusReport.stopStatusReportSchedule()
         piscinaStatsJob && schedule.cancelJob(piscinaStatsJob)
         internalMetricsStatsJob && schedule.cancelJob(internalMetricsStatsJob)
+        flushLastSeenAtCacheJob && schedule.cancelJob(flushLastSeenAtCacheJob)
+        flushEventPropertyCounterJob && schedule.cancelJob(flushEventPropertyCounterJob)
         await jobQueueConsumer?.stop()
         await scheduleControl?.stopSchedule()
         await new Promise<void>((resolve, reject) =>
@@ -198,10 +201,17 @@ export async function startPluginsServer(
             })
         }
 
-        // every minute flush lastSeenAt cache
+        // every 10 seconds past the minute flush lastSeenAt cache
         if (serverConfig.EXPERIMENTAL_EVENTS_LAST_SEEN_ENABLED) {
-            flushLastSeenAtCacheJob = schedule.scheduleJob('0 * * * * *', async () => {
-                await hub!.teamManager.flushLastSeenAtCache()
+            flushLastSeenAtCacheJob = schedule.scheduleJob('10 * * * * *', async () => {
+                await piscina!.broadcastTask({ task: 'flushLastSeenAtCache' })
+            })
+        }
+
+        // every 20 seconds past the minute flush the event property counter cache
+        if (serverConfig.EXPERIMENTAL_EVENT_PROPERTY_COUNTER_ENABLED_TEAMS) {
+            flushEventPropertyCounterJob = schedule.scheduleJob('20 * * * * *', async () => {
+                await piscina!.broadcastTask({ task: 'flushEventPropertyCounter' })
             })
         }
 
@@ -268,6 +278,7 @@ export async function stopPiscina(piscina: Piscina): Promise<void> {
     await Promise.all([
         piscina.broadcastTask({ task: 'flushKafkaMessages' }),
         piscina.broadcastTask({ task: 'flushLastSeenAtCache' }),
+        piscina.broadcastTask({ task: 'flushEventPropertyCounter' }),
         delay(2000),
     ])
     await piscina.destroy()

plugin-server/src/types.ts

@@ -17,6 +17,7 @@ import { PluginMetricsManager } from './utils/plugin-metrics'
 import { UUID } from './utils/utils'
 import { ActionManager } from './worker/ingestion/action-manager'
 import { ActionMatcher } from './worker/ingestion/action-matcher'
+import { EventPropertyCounter } from './worker/ingestion/event-property-counter'
 import { HookCommander } from './worker/ingestion/hooks'
 import { OrganizationManager } from './worker/ingestion/organization-manager'
 import { EventsProcessor } from './worker/ingestion/process-event'
@@ -104,6 +105,7 @@ export interface PluginsServerConfig extends Record<string, any> {
     SITE_URL: string | null
     NEW_PERSON_PROPERTIES_UPDATE_ENABLED_TEAMS: string
     EXPERIMENTAL_EVENTS_LAST_SEEN_ENABLED: boolean
+    EXPERIMENTAL_EVENT_PROPERTY_COUNTER_ENABLED_TEAMS: string
 }
 
 export interface Hub extends PluginsServerConfig {
@@ -131,6 +133,7 @@ export interface Hub extends PluginsServerConfig {
     pluginConfigSecretLookup: Map<string, PluginConfigId>
     // tools
     teamManager: TeamManager
+    eventPropertyCounter: EventPropertyCounter
     organizationManager: OrganizationManager
     pluginsApiKeyManager: PluginsApiKeyManager
     actionManager: ActionManager
@@ -761,6 +764,18 @@ export interface PropertyDefinitionType {
     team_id: number
 }
 
+export interface EventPropertyType {
+    id: string
+    event: string
+    property: string
+    property_type: string | null
+    property_type_format: string | null
+    total_volume: number | null
+    created_at: string // DateTime
+    last_seen_at: string // DateTime
+    team_id: number
+}
+
 export type PluginFunction = 'onEvent' | 'onAction' | 'processEvent' | 'onSnapshot' | 'pluginTask'
 
 export enum CeleryTriggeredJobOperation {

plugin-server/src/utils/db/db.ts

@@ -31,6 +31,7 @@ import {
     ElementGroup,
     Event,
     EventDefinitionType,
+    EventPropertyType,
     Group,
     GroupTypeIndex,
     GroupTypeToColumnIndex,
@@ -1119,6 +1120,13 @@ export class DB {
         ).rows as PropertyDefinitionType[]
     }
 
+    // EventProperty
+
+    public async fetchEventProperties(): Promise<EventPropertyType[]> {
+        return (await this.postgresQuery('SELECT * FROM posthog_eventproperty', undefined, 'fetchEventProperties'))
+            .rows as EventPropertyType[]
+    }
+
     // Action & ActionStep & Action<>Event
 
     public async fetchAllActionsGroupedByTeam(): Promise<Record<Team['id'], Record<Action['id'], Action>>> {

plugin-server/src/utils/db/hub.ts

@@ -15,6 +15,7 @@ import { JobQueueManager } from '../../main/job-queues/job-queue-manager'
 import { Hub, PluginsServerConfig } from '../../types'
 import { ActionManager } from '../../worker/ingestion/action-manager'
 import { ActionMatcher } from '../../worker/ingestion/action-matcher'
+import { EventPropertyCounter } from '../../worker/ingestion/event-property-counter'
 import { HookCommander } from '../../worker/ingestion/hooks'
 import { OrganizationManager } from '../../worker/ingestion/organization-manager'
 import { EventsProcessor } from '../../worker/ingestion/process-event'
@@ -165,6 +166,7 @@ export async function createHub(
         serverConfig.SITE_URL,
         serverConfig.EXPERIMENTAL_EVENTS_LAST_SEEN_ENABLED
     )
+    const eventPropertyCounter = new EventPropertyCounter(db, statsd)
     const organizationManager = new OrganizationManager(db)
     const pluginsApiKeyManager = new PluginsApiKeyManager(db)
     const actionManager = new ActionManager(db)
@@ -191,6 +193,7 @@ export async function createHub(
         pluginSchedulePromises: { runEveryMinute: {}, runEveryHour: {}, runEveryDay: {} },
 
         teamManager,
+        eventPropertyCounter,
         organizationManager,
         pluginsApiKeyManager,
         actionManager,

plugin-server/src/utils/db/postgres-logs-wrapper.ts

@@ -36,8 +36,9 @@ export class PostgresLogsWrapper {
             this.flushTimeout = null
         }
         if (this.logs.length > 0) {
-            await this.db.batchInsertPostgresLogs(this.logs)
+            const logs = this.logs
             this.logs = []
+            await this.db.batchInsertPostgresLogs(logs)
         }
     }
 }

plugin-server/src/worker/ingestion/event-property-counter.ts

@@ -0,0 +1,178 @@
+import { Properties } from '@posthog/plugin-scaffold'
+import { StatsD } from 'hot-shots'
+import { DateTime } from 'luxon'
+
+import { TeamId } from '../../types'
+import { DB } from '../../utils/db/db'
+import { timeoutGuard } from '../../utils/db/utils'
+import { status } from '../../utils/status'
+
+enum EventPropertyType {
+    Number = 'NUMBER',
+    String = 'STRING',
+    Boolean = 'BOOLEAN',
+    DateTime = 'DATETIME',
+}
+
+interface EventPropertiesCounter {
+    createdAt: number
+    lastSeenAt: number
+    totalVolume: number
+    propertyType: EventPropertyType | null
+    propertyTypeFormat: string | null
+}
+
+interface EventNamePropertiesBuffer {
+    totalVolume: number
+    uniqueVolume: number
+    buffer: Map<TeamId, Map<string, EventPropertiesCounter>>
+}
+
+export class EventPropertyCounter {
+    db: DB
+    eventPropertiesBuffer: EventNamePropertiesBuffer
+    lastFlushAt: DateTime
+    statsd?: StatsD
+
+    constructor(db: DB, statsd?: StatsD) {
+        this.db = db
+        this.statsd = statsd
+        this.eventPropertiesBuffer = { totalVolume: 0, uniqueVolume: 0, buffer: new Map() }
+        this.lastFlushAt = DateTime.now()
+    }
+
+    public async updateEventPropertyCounter(teamId: number, event: string, properties: Properties): Promise<void> {
+        this.updateEventPropertiesBuffer(teamId, event, properties)
+        // Flush every 2 minutes or 50k unique properties, whichever comes first.
+        // Additionally, a flush is broadcast every 1 minute from pluginServer.
+        if (
+            this.eventPropertiesBuffer.uniqueVolume > 50000 ||
+            DateTime.now().diff(this.lastFlushAt).as('seconds') > 120
+        ) {
+            await this.flush()
+        }
+    }
+
+    /** Save information about event properties into a custom buffer */
+    public updateEventPropertiesBuffer(teamId: number, event: string, properties: Record<string, any>): void {
+        const timestamp = DateTime.now().toSeconds()
+        let bufferForTeam = this.eventPropertiesBuffer.buffer.get(teamId)
+        if (!bufferForTeam) {
+            bufferForTeam = new Map()
+            this.eventPropertiesBuffer.buffer.set(teamId, bufferForTeam)
+        }
+        for (const [property, value] of Object.entries(properties)) {
+            const key = JSON.stringify([event, property])
+            let propertyBuffer = bufferForTeam.get(key)
+            if (!propertyBuffer) {
+                propertyBuffer = {
+                    createdAt: timestamp,
+                    lastSeenAt: timestamp,
+                    totalVolume: 1,
+                    propertyType: null,
+                    propertyTypeFormat: null,
+                }
+                bufferForTeam.set(key, propertyBuffer)
+                this.eventPropertiesBuffer.uniqueVolume += 1
+            } else {
+                propertyBuffer.createdAt = Math.min(timestamp, propertyBuffer.createdAt)
+                propertyBuffer.lastSeenAt = Math.max(timestamp, propertyBuffer.lastSeenAt)
+                propertyBuffer.totalVolume += 1
+            }
+            this.eventPropertiesBuffer.totalVolume += 1
+
+            propertyBuffer.propertyType =
+                typeof value === 'number'
+                    ? EventPropertyType.Number
+                    : typeof value === 'boolean'
+                    ? EventPropertyType.Boolean
+                    : typeof value === 'string'
+                    ? EventPropertyType.String
+                    : null
+            propertyBuffer.propertyTypeFormat = null
+
+            if (propertyBuffer.propertyType === EventPropertyType.String) {
+                const dateFormat = detectDateFormat(value)
+                if (dateFormat) {
+                    propertyBuffer.propertyType = EventPropertyType.DateTime
+                    propertyBuffer.propertyTypeFormat = dateFormat
+                }
+            }
+        }
+    }
+
+    public async flush(): Promise<void> {
+        if (this.eventPropertiesBuffer.totalVolume === 0) {
+            return
+        }
+        const timeout = timeoutGuard(
+            `Still flushing the event names and properties buffer. Timeout warning after 30 sec!`
+        )
+        try {
+            const startTime = DateTime.now()
+            const lastFlushedSecondsAgo = DateTime.now().diff(this.lastFlushAt).as('seconds')
+            const cacheSize = this.eventPropertiesBuffer.uniqueVolume
+            status.info(
+                `🚽 Starting flushEventPropertyCounter. Cache size: ${cacheSize} items. Last flushed: ${lastFlushedSecondsAgo} seconds ago.`
+            )
+
+            const oldBuffer = this.eventPropertiesBuffer
+            this.eventPropertiesBuffer = { totalVolume: 0, uniqueVolume: 0, buffer: new Map() }
+            this.lastFlushAt = DateTime.now()
+
+            const columns: any[][] = [[], [], [], [], [], [], [], []]
+
+            for (const [teamId, teamBuffer] of oldBuffer.buffer.entries()) {
+                for (const [key, propertyBuffer] of teamBuffer.entries()) {
+                    const [event, property] = JSON.parse(key)
+                    const { propertyType, propertyTypeFormat, totalVolume, lastSeenAt, createdAt } = propertyBuffer
+                    columns[0].push(teamId)
+                    columns[1].push(event)
+                    columns[2].push(property)
+                    columns[3].push(propertyType)
+                    columns[4].push(propertyTypeFormat)
+                    columns[5].push(totalVolume)
+                    columns[6].push(DateTime.fromSeconds(createdAt).toUTC())
+                    columns[7].push(DateTime.fromSeconds(lastSeenAt).toUTC())
+                }
+            }
+
+            // VALUES ${queryValues.join(',')}
+            await this.db.postgresQuery(
+                `INSERT INTO posthog_eventproperty(team_id, event, property, property_type, property_type_format, total_volume, created_at, last_seen_at)
+                SELECT * FROM UNNEST ($1::int[], $2::text[], $3::text[], $4::text[], $5::text[], $6::bigint[], $7::timestamp with time zone[], $8::timestamp with time zone[])
+                ON CONFLICT ON CONSTRAINT posthog_eventproperty_team_id_event_property_10910b3b_uniq DO UPDATE SET
+                    total_volume = posthog_eventproperty.total_volume + excluded.total_volume,
+                    created_at = LEAST(posthog_eventproperty.created_at, excluded.created_at),
+                    last_seen_at = GREATEST(posthog_eventproperty.last_seen_at, excluded.last_seen_at),
+                    property_type = CASE WHEN posthog_eventproperty.property_type IS NULL THEN excluded.property_type ELSE posthog_eventproperty.property_type END,
+                    property_type_format = CASE WHEN posthog_eventproperty.property_type_format IS NULL THEN excluded.property_type_format ELSE posthog_eventproperty.property_type_format END
+                `,
+                columns,
+                'eventPropertyCounterFlush'
+            )
+
+            const elapsedTime = DateTime.now().diff(startTime).as('milliseconds')
+            this.statsd?.set('flushEventPropertyCounter.Size', cacheSize)
+            this.statsd?.set('flushEventPropertyCounter.QuerySize', columns[0].length)
+            this.statsd?.timing('flushEventPropertyCounter', elapsedTime)
+            status.info(`✅ 🚽 flushEventPropertyCounter finished successfully in ${elapsedTime} ms.`)
+        } finally {
+            clearTimeout(timeout)
+        }
+    }
+}
+
+export function detectDateFormat(value: string): string | void {
+    if (value.match(/^\d{4}-\d{2}-\d{2}$/)) {
+        return 'YYYY-MM-DD'
+    }
+
+    if (value.match(/^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(.\d{1,3}?Z?|)$/)) {
+        return 'ISO8601 UTC'
+    }
+
+    if (value.match(/^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(.\d{1,3}|)\+\d{2}:\d{2}$/)) {
+        return 'ISO8601 TZ'
+    }
+}