From 42eb852a27748ca9e1966c78891b66f9b363ea2b Mon Sep 17 00:00:00 2001 From: Michael Matloka Date: Fri, 21 May 2021 03:53:22 +0200 Subject: [PATCH 01/21] Add ActionManager --- benchmarks/postgres/ingestion.benchmark.ts | 2 + src/main/pluginsServer.ts | 20 ++-- src/types.ts | 110 ++++++++++++++++++ src/utils/db/db.ts | 62 +++++++++- src/utils/db/server.ts | 2 + src/utils/pubsub.ts | 50 ++++++++ src/utils/utils.ts | 2 +- src/worker/ingestion/action-manager.ts | 74 ++++++++++++ src/worker/ingestion/process-event.ts | 17 +++ tests/helpers/sql.ts | 44 ++++++- tests/shared/process-event.ts | 2 + tests/worker/ingestion/action-manager.test.ts | 86 ++++++++++++++ 12 files changed, 456 insertions(+), 15 deletions(-) create mode 100644 src/utils/pubsub.ts create mode 100644 src/worker/ingestion/action-manager.ts create mode 100644 tests/worker/ingestion/action-manager.test.ts diff --git a/benchmarks/postgres/ingestion.benchmark.ts b/benchmarks/postgres/ingestion.benchmark.ts index cd5c7caf..bc6319b4 100644 --- a/benchmarks/postgres/ingestion.benchmark.ts +++ b/benchmarks/postgres/ingestion.benchmark.ts @@ -52,6 +52,7 @@ describe('ingestion benchmarks', () => { LOG_LEVEL: LogLevel.Log, }) eventsProcessor = new EventsProcessor(server) + await eventsProcessor.prepare() team = await getFirstTeam(server) now = DateTime.utc() @@ -62,6 +63,7 @@ describe('ingestion benchmarks', () => { }) afterEach(async () => { + await eventsProcessor.close() await stopServer?.() }) diff --git a/src/main/pluginsServer.ts b/src/main/pluginsServer.ts index 68408830..9c873c57 100644 --- a/src/main/pluginsServer.ts +++ b/src/main/pluginsServer.ts @@ -2,7 +2,6 @@ import { ReaderModel } from '@maxmind/geoip2-node' import Piscina from '@posthog/piscina' import * as Sentry from '@sentry/node' import { FastifyInstance } from 'fastify' -import Redis from 'ioredis' import net, { AddressInfo } from 'net' import * as schedule from 'node-schedule' @@ -10,8 +9,9 @@ import { defaultConfig } from '../config/config' import { JobQueueConsumerControl, PluginsServer, PluginsServerConfig, Queue, ScheduleControl } from '../types' import { createServer } from '../utils/db/server' import { killProcess } from '../utils/kill' +import { PubSub } from '../utils/pubsub' import { status } from '../utils/status' -import { createRedis, delay, getPiscinaStats } from '../utils/utils' +import { delay, getPiscinaStats } from '../utils/utils' import { startQueue } from './ingestion-queues/queue' import { startJobQueueConsumer } from './job-queues/job-queue-consumer' import { createMmdbServer, performMmdbStalenessCheck, prepareMmdb } from './services/mmdb' @@ -41,7 +41,7 @@ export async function startPluginsServer( status.info('ℹī¸', `${serverConfig.WORKER_CONCURRENCY} workers, ${serverConfig.TASKS_PER_WORKER} tasks per worker`) - let pubSub: Redis.Redis | undefined + let pubSub: PubSub | undefined let server: PluginsServer | undefined let fastifyInstance: FastifyInstance | undefined let pingJob: schedule.Job | undefined @@ -73,7 +73,7 @@ export async function startPluginsServer( } lastActivityCheck && clearInterval(lastActivityCheck) await queue?.stop() - await pubSub?.quit() + await pubSub?.stop() pingJob && schedule.cancelJob(pingJob) statsJob && schedule.cancelJob(statsJob) await jobQueueConsumer?.stop() @@ -141,17 +141,15 @@ export async function startPluginsServer( void jobQueueConsumer?.resume() }) - // use one extra connection for redis pubsub - pubSub = await createRedis(server) - await pubSub.subscribe(server.PLUGINS_RELOAD_PUBSUB_CHANNEL) - pubSub.on('message', async (channel: string, message) => { - if (channel === server!.PLUGINS_RELOAD_PUBSUB_CHANNEL) { + // use one extra connection for Redis-based PubSub + pubSub = new PubSub(server, { + [server.PLUGINS_RELOAD_PUBSUB_CHANNEL]: async () => { status.info('⚡', 'Reloading plugins!') - await piscina?.broadcastTask({ task: 'reloadPlugins' }) await scheduleControl?.reloadSchedule() - } + }, }) + await pubSub.start() if (server.jobQueueManager) { const queueString = server.jobQueueManager.getJobQueueTypesAsString() diff --git a/src/types.ts b/src/types.ts index 5b040109..3edd5926 100644 --- a/src/types.ts +++ b/src/types.ts @@ -312,6 +312,7 @@ export interface RawOrganization { name: string created_at: string updated_at: string + available_features: string[] } /** Usable Team model. */ @@ -422,6 +423,115 @@ export interface CohortPeople { person_id: number } +/** Raw Action row from database. */ +export interface RawAction { + id: number + team_id: TeamId + name: string | null + created_at: string + created_by_id: number | null + deleted: boolean + post_to_slack: boolean + slack_message_format: string + is_calculating: boolean + updated_at: string + last_calculated_at: string +} + +/** Sync with posthog/frontend/src/types.ts */ +export enum PropertyOperator { + Exact = 'exact', + IsNot = 'is_not', + IContains = 'icontains', + NotIContains = 'not_icontains', + Regex = 'regex', + NotRegex = 'not_regex', + GreaterThan = 'gt', + LessThan = 'lt', + IsSet = 'is_set', + IsNotSet = 'is_not_set', +} + +/** Sync with posthog/frontend/src/types.ts */ +interface BasePropertyFilter { + key: string + value: string | number | Array | null + label?: string +} + +/** Sync with posthog/frontend/src/types.ts */ +export interface EventPropertyFilter extends BasePropertyFilter { + type: 'event' + operator: PropertyOperator +} + +/** Sync with posthog/frontend/src/types.ts */ +export interface PersonPropertyFilter extends BasePropertyFilter { + type: 'person' + operator: PropertyOperator +} + +/** Sync with posthog/frontend/src/types.ts */ +export interface ElementPropertyFilter extends BasePropertyFilter { + type: 'element' + key: 'tag_name' | 'text' | 'href' | 'selector' + operator: PropertyOperator +} + +/** Sync with posthog/frontend/src/types.ts */ +export interface CohortPropertyFilter extends BasePropertyFilter { + type: 'cohort' + key: 'id' + value: number +} + +/** Sync with posthog/frontend/src/types.ts */ +export type ActionStepProperties = + | EventPropertyFilter + | PersonPropertyFilter + | ElementPropertyFilter + | CohortPropertyFilter + +/** Sync with posthog/frontend/src/types.ts */ +export enum ActionStepUrlMatching { + Contains = 'contains', + Regex = 'regex', + Exact = 'exact', +} + +export interface ActionStep { + id: number + action_id: number + tag_name: string | null + text: string | null + href: string | null + selector: string | null + url: string | null + url_matching: ActionStepUrlMatching | null + name: string | null + event: string | null + properties: ActionStepProperties[] | null +} + +/** Raw Action row from database. */ +export interface RawAction { + id: number + name: string | null + created_at: string + created_by_id: number | null + deleted: boolean + post_to_slack: boolean + slack_message_format: string + is_calculating: boolean + updated_at: string + last_calculated_at: string +} + +/** Usable Action model. */ +export interface Action extends RawAction { + steps: ActionStep[] +} + export interface SessionRecordingEvent { uuid: string timestamp: string diff --git a/src/utils/db/db.ts b/src/utils/db/db.ts index c00f1a54..568457c9 100644 --- a/src/utils/db/db.ts +++ b/src/utils/db/db.ts @@ -10,6 +10,8 @@ import { Pool, PoolClient, QueryConfig, QueryResult, QueryResultRow } from 'pg' import { KAFKA_PERSON, KAFKA_PERSON_UNIQUE_ID, KAFKA_PLUGIN_LOG_ENTRIES } from '../../config/kafka-topics' import { + Action, + ActionStep, ClickHouseEvent, ClickHousePerson, ClickHousePersonDistinctId, @@ -26,6 +28,7 @@ import { PluginLogEntryType, PostgresSessionRecordingEvent, PropertyDefinitionType, + RawAction, RawOrganization, RawPerson, SessionRecordingEvent, @@ -398,7 +401,7 @@ export class DB { return `|| CASE WHEN (COALESCE(properties->>'${sanitizedPropName}', '0')~E'^([-+])?[0-9\.]+$') THEN jsonb_build_object('${sanitizedPropName}', (COALESCE(properties->>'${sanitizedPropName}','0')::numeric + $${ index + 1 - })) + })) ELSE '{}' END ` }) @@ -682,14 +685,71 @@ export class DB { return entry } + // EventDefinition + public async fetchEventDefinitions(): Promise { return (await this.postgresQuery('SELECT * FROM posthog_eventdefinition', undefined, 'fetchEventDefinitions')) .rows as EventDefinitionType[] } + // PropertyDefinition + public async fetchPropertyDefinitions(): Promise { return ( await this.postgresQuery('SELECT * FROM posthog_propertydefinition', undefined, 'fetchPropertyDefinitions') ).rows as PropertyDefinitionType[] } + + // Action & ActionStep + + public async fetchAllActionsMap(): Promise> { + const rawActions: RawAction[] = ( + await this.postgresQuery( + ` + SELECT * FROM posthog_action`, + undefined, + 'fetchActions' + ) + ).rows + const actionSteps: ActionStep[] = ( + await this.postgresQuery( + ` + SELECT * FROM posthog_actionstep`, + undefined, + 'fetchActionSteps' + ) + ).rows + const actionsMap: Record = {} + for (const rawAction of rawActions) { + actionsMap[rawAction.id] = { ...rawAction, steps: [] } + } + for (const actionStep of actionSteps) { + actionsMap[actionStep.action_id].steps.push(actionStep) + } + return actionsMap + } + + public async fetchAction(id: Action['id']): Promise { + const rawActions: RawAction[] = ( + await this.postgresQuery( + ` + SELECT * FROM posthog_action WHERE id = $1`, + [id], + 'fetchActions' + ) + ).rows + if (!rawActions.length) { + return null + } + const steps: ActionStep[] = ( + await this.postgresQuery( + ` + SELECT * FROM posthog_actionstep WHERE action_id = $1`, + [id], + 'fetchActionSteps' + ) + ).rows + const action: Action = { ...rawActions[0], steps } + return action + } } diff --git a/src/utils/db/server.ts b/src/utils/db/server.ts index c0eba78b..1dc4691b 100644 --- a/src/utils/db/server.ts +++ b/src/utils/db/server.ts @@ -175,6 +175,7 @@ export async function createServer( // :TODO: This is only used on worker threads, not main server.eventsProcessor = new EventsProcessor(server as PluginsServer) + await server.eventsProcessor.prepare() server.jobQueueManager = new JobQueueManager(server as PluginsServer) try { @@ -192,6 +193,7 @@ export async function createServer( clearInterval(eventLoopLagInterval) } server.mmdbUpdateJob?.cancel() + await server.eventsProcessor?.close() await server.jobQueueManager?.disconnectProducer() if (kafkaProducer) { clearInterval(kafkaProducer.flushInterval) diff --git a/src/utils/pubsub.ts b/src/utils/pubsub.ts new file mode 100644 index 00000000..5797aac8 --- /dev/null +++ b/src/utils/pubsub.ts @@ -0,0 +1,50 @@ +import { Redis } from 'ioredis' + +import { PluginsServerConfig } from '../types' +import { createRedis } from './utils' + +export type PubSubTask = ((message: string) => void) | ((message: string) => Promise) + +export interface PubSubTaskMap { + [channel: string]: PubSubTask +} + +export class PubSub { + private serverConfig: PluginsServerConfig + private redis: Redis | null + public taskMap: PubSubTaskMap + + constructor(serverConfig: PluginsServerConfig, taskMap: PubSubTaskMap = {}) { + this.serverConfig = serverConfig + this.redis = null + this.taskMap = taskMap + } + + public async start(): Promise { + if (this.redis) { + throw new Error('Started PubSub cannot be started again!') + } + this.redis = await createRedis(this.serverConfig) + await this.redis.subscribe(Object.keys(this.taskMap)) + this.redis.on('message', (channel: string, message: string) => { + const task: PubSubTask | undefined = this.taskMap[channel] + if (!task) { + throw new Error( + `Received a pubsub message for unassociated channel ${channel}! Associated channels are: ${Object.keys( + this.taskMap + )}` + ) + } + void task(message) + }) + } + + public async stop(): Promise { + if (!this.redis) { + throw new Error('Unstarted PubSub cannot be stopped!') + } + await this.redis.unsubscribe() + this.redis.disconnect() + this.redis = null + } +} diff --git a/src/utils/utils.ts b/src/utils/utils.ts index 7019c290..536b0689 100644 --- a/src/utils/utils.ts +++ b/src/utils/utils.ts @@ -5,7 +5,7 @@ import AdmZip from 'adm-zip' import { randomBytes } from 'crypto' import Redis, { RedisOptions } from 'ioredis' import { DateTime } from 'luxon' -import { Pool, PoolClient, PoolConfig } from 'pg' +import { Pool, PoolConfig } from 'pg' import { Readable } from 'stream' import * as tar from 'tar-stream' import * as zlib from 'zlib' diff --git a/src/worker/ingestion/action-manager.ts b/src/worker/ingestion/action-manager.ts new file mode 100644 index 00000000..df9d996d --- /dev/null +++ b/src/worker/ingestion/action-manager.ts @@ -0,0 +1,74 @@ +import { Action,PluginsServerConfig } from '../../types' +import { DB } from '../../utils/db/db' +import { PubSub } from '../../utils/pubsub' +import { status } from '../../utils/status' + +type ActionCache = Record + +export class ActionManager { + private ready: boolean + private db: DB + public pubSub: PubSub + private actionCache: ActionCache + + constructor(db: DB, serverConfig: PluginsServerConfig) { + this.ready = false + this.db = db + this.pubSub = new PubSub(serverConfig, { + 'fetch-action': async (message) => { + const actionId = parseInt(message) + const refetchedAction = await this.db.fetchAction(actionId) + if (refetchedAction) { + status.info( + 'đŸŋ', + actionId in this.actionCache + ? `Refetched action ID ${actionId} from DB` + : `Fetched new action ID ${actionId} from DB` + ) + this.actionCache[actionId] = refetchedAction + } else if (actionId in this.actionCache) { + status.info( + 'đŸŋ', + `Tried to fetch action ID ${actionId} from DB, but it wasn't found in DB, so deleted from cache instead` + ) + delete this.actionCache[actionId] + } else { + status.info( + 'đŸŋ', + `Tried to fetch action ID ${actionId} from DB, but it wasn't found in DB or cache, so did nothing instead` + ) + } + }, + 'delete-action': (message) => { + const actionId = parseInt(message) + if (actionId in this.actionCache) { + status.info('đŸŋ', `Deleted action ID ${actionId} from cache`) + delete this.actionCache[actionId] + } else { + status.info( + 'đŸŋ', + `Tried to delete action ID ${actionId} from cache, but it wasn't found in cache, so did nothing instead` + ) + } + }, + }) + this.actionCache = {} + } + + public async prepare(): Promise { + this.actionCache = await this.db.fetchAllActionsMap() + await this.pubSub.start() + this.ready = true + } + + public async close(): Promise { + await this.pubSub.stop() + } + + public getAction(id: Action['id']): Action | undefined { + if (!this.ready) { + throw new Error('ActionManager is not ready! Run actionManager.prepare() before this') + } + return this.actionCache[id] + } +} diff --git a/src/worker/ingestion/process-event.ts b/src/worker/ingestion/process-event.ts index b8940139..3b2142f5 100644 --- a/src/worker/ingestion/process-event.ts +++ b/src/worker/ingestion/process-event.ts @@ -26,10 +26,12 @@ import { KafkaProducerWrapper } from '../../utils/db/kafka-producer-wrapper' import { elementsToString, personInitialAndUTMProperties, sanitizeEventName, timeoutGuard } from '../../utils/db/utils' import { status } from '../../utils/status' import { castTimestampOrNow, filterIncrementProperties, UUID, UUIDT } from '../../utils/utils' +import { ActionManager } from './action-manager' import { PersonManager } from './person-manager' import { TeamManager } from './team-manager' export class EventsProcessor { + ready: boolean pluginsServer: PluginsServer db: DB clickhouse: ClickHouse | undefined @@ -38,8 +40,10 @@ export class EventsProcessor { posthog: ReturnType teamManager: TeamManager personManager: PersonManager + actionManager: ActionManager constructor(pluginsServer: PluginsServer) { + this.ready = false this.pluginsServer = pluginsServer this.db = pluginsServer.db this.clickhouse = pluginsServer.clickhouse @@ -47,6 +51,7 @@ export class EventsProcessor { this.celery = new Client(pluginsServer.db, pluginsServer.CELERY_DEFAULT_QUEUE) this.teamManager = new TeamManager(pluginsServer.db) this.personManager = new PersonManager(pluginsServer) + this.actionManager = new ActionManager(pluginsServer.db, pluginsServer) this.posthog = nodePostHog('sTMFPsFhdP1Ssg', { fetch }) if (process.env.NODE_ENV === 'test') { @@ -54,6 +59,15 @@ export class EventsProcessor { } } + public async prepare(): Promise { + await this.actionManager.prepare() + this.ready = true + } + + public async close(): Promise { + await this.actionManager.close() + } + public async processEvent( distinctId: string, ip: string | null, @@ -64,6 +78,9 @@ export class EventsProcessor { sentAt: DateTime | null, eventUuid: string ): Promise { + if (!this.ready) { + throw new Error('EventsProcessor is not ready! Run eventsProcessor.prepare() before this') + } if (!UUID.validateString(eventUuid, false)) { throw new Error(`Not a valid UUID: "${eventUuid}"`) } diff --git a/tests/helpers/sql.ts b/tests/helpers/sql.ts index 8b0d8a4c..1f360d57 100644 --- a/tests/helpers/sql.ts +++ b/tests/helpers/sql.ts @@ -1,7 +1,18 @@ import { Pool, PoolClient } from 'pg' import { defaultConfig } from '../../src/config/config' -import { Plugin, PluginAttachmentDB, PluginConfig, PluginsServer, PluginsServerConfig, Team } from '../../src/types' +import { + ActionStep, + Plugin, + PluginAttachmentDB, + PluginConfig, + PluginsServer, + PluginsServerConfig, + PropertyOperator, + RawAction, + RawOrganization, + Team, +} from '../../src/types' import { UUIDT } from '../../src/utils/utils' import { commonOrganizationId, @@ -29,6 +40,8 @@ export async function resetTestDatabase( } catch {} await db.query(` + DELETE FROM posthog_actionstep; + DELETE FROM posthog_action; DELETE FROM posthog_element; DELETE FROM posthog_elementgroup; DELETE FROM posthog_sessionrecordingevent; @@ -109,7 +122,8 @@ export async function createUserTeamAndOrganization( personalization: '{}', setup_section_2_completed: true, for_internal_metrics: false, - }) + available_features: [], + } as RawOrganization) await insertRow(db, 'posthog_organizationmembership', { id: organizationMembershipId, organization_id: organizationId, @@ -143,6 +157,32 @@ export async function createUserTeamAndOrganization( timezone: 'UTC', data_attributes: JSON.stringify(['data-attr']), }) + await insertRow(db, 'posthog_action', { + id: 67, + team_id: teamId, + name: 'Test Action', + created_at: new Date().toISOString(), + created_by_id: userId, + deleted: false, + post_to_slack: false, + slack_message_format: '', + is_calculating: false, + updated_at: new Date().toISOString(), + last_calculated_at: new Date().toISOString(), + } as RawAction) + await insertRow(db, 'posthog_actionstep', { + id: 911, + action_id: 67, + tag_name: null, + text: null, + href: null, + selector: null, + url: null, + url_matching: null, + name: null, + event: null, + properties: JSON.stringify([{ type: 'event', operator: PropertyOperator.Exact, key: 'foo', value: ['bar'] }]), + }) } export async function getTeams(server: PluginsServer): Promise { diff --git a/tests/shared/process-event.ts b/tests/shared/process-event.ts index 1f0f7602..d34aaf68 100644 --- a/tests/shared/process-event.ts +++ b/tests/shared/process-event.ts @@ -128,6 +128,7 @@ export const createProcessEventTests = ( returned.server = server returned.stopServer = stopServer eventsProcessor = new EventsProcessor(server) + await eventsProcessor.prepare() queryCounter = 0 processEventCounter = 0 team = await getFirstTeam(server) @@ -139,6 +140,7 @@ export const createProcessEventTests = ( }) afterEach(async () => { + await eventsProcessor.close() await server.redisPool.release(redis) await stopServer?.() }) diff --git a/tests/worker/ingestion/action-manager.test.ts b/tests/worker/ingestion/action-manager.test.ts new file mode 100644 index 00000000..178eee31 --- /dev/null +++ b/tests/worker/ingestion/action-manager.test.ts @@ -0,0 +1,86 @@ +import { PluginsServer, PropertyOperator } from '../../../src/types' +import { createServer } from '../../../src/utils/db/server' +import { createRedis, delay } from '../../../src/utils/utils' +import { ActionManager } from '../../../src/worker/ingestion/action-manager' +import { resetTestDatabase } from '../../helpers/sql' + +describe('ActionManager()', () => { + let server: PluginsServer + let closeServer: () => Promise + let actionManager: ActionManager + + beforeEach(async () => { + ;[server, closeServer] = await createServer() + await resetTestDatabase() + actionManager = new ActionManager(server.db, server) + await actionManager.prepare() + }) + afterEach(async () => { + await actionManager.close() + await closeServer() + }) + + describe('getAction()', () => { + it('returns the correct action', async () => { + const action = actionManager.getAction(67) + + expect(action).toMatchObject({ + id: 67, + name: 'Test Action', + deleted: false, + post_to_slack: false, + slack_message_format: '', + is_calculating: false, + steps: [ + { + id: 911, + action_id: 67, + tag_name: null, + text: null, + href: null, + selector: null, + url: null, + url_matching: null, + name: null, + event: null, + properties: [{ type: 'event', operator: PropertyOperator.Exact, key: 'foo', value: ['bar'] }], + }, + ], + }) + + await server.db.postgresQuery( + `UPDATE posthog_actionstep SET properties = jsonb_set(properties, '{0,key}', '"baz"') WHERE id = 911`, + undefined, + 'testKey' + ) + + // This is normally done by Django async in such a situation + await (actionManager.pubSub.taskMap['fetch-action']('67') as Promise) + const reloadedAction = actionManager.getAction(67) + + expect(reloadedAction).toMatchObject({ + id: 67, + name: 'Test Action', + deleted: false, + post_to_slack: false, + slack_message_format: '', + is_calculating: false, + steps: [ + { + id: 911, + action_id: 67, + tag_name: null, + text: null, + href: null, + selector: null, + url: null, + url_matching: null, + name: null, + event: null, + properties: [{ type: 'event', operator: PropertyOperator.Exact, key: 'baz', value: ['bar'] }], + }, + ], + }) + }) + }) +}) From bb4501096b513a9a9bd49ca03b9954779e72ec01 Mon Sep 17 00:00:00 2001 From: Michael Matloka Date: Fri, 21 May 2021 04:14:59 +0200 Subject: [PATCH 02/21] Refactor ActionManager --- src/utils/pubsub.ts | 8 +- src/worker/ingestion/action-manager.ts | 83 ++++++++----------- src/worker/ingestion/process-event.ts | 11 ++- tests/worker/ingestion/action-manager.test.ts | 5 +- 4 files changed, 53 insertions(+), 54 deletions(-) diff --git a/src/utils/pubsub.ts b/src/utils/pubsub.ts index 5797aac8..0884b07b 100644 --- a/src/utils/pubsub.ts +++ b/src/utils/pubsub.ts @@ -1,6 +1,7 @@ import { Redis } from 'ioredis' import { PluginsServerConfig } from '../types' +import { status } from './status' import { createRedis } from './utils' export type PubSubTask = ((message: string) => void) | ((message: string) => Promise) @@ -25,18 +26,20 @@ export class PubSub { throw new Error('Started PubSub cannot be started again!') } this.redis = await createRedis(this.serverConfig) - await this.redis.subscribe(Object.keys(this.taskMap)) + const channels = Object.keys(this.taskMap) + await this.redis.subscribe(channels) this.redis.on('message', (channel: string, message: string) => { const task: PubSubTask | undefined = this.taskMap[channel] if (!task) { throw new Error( `Received a pubsub message for unassociated channel ${channel}! Associated channels are: ${Object.keys( this.taskMap - )}` + ).join(', ')}` ) } void task(message) }) + status.info('👀', `Pub-sub started for channels: ${channels.join(', ')}`) } public async stop(): Promise { @@ -46,5 +49,6 @@ export class PubSub { await this.redis.unsubscribe() this.redis.disconnect() this.redis = null + status.info('🛑', `Pub-sub stopped for channels: ${Object.keys(this.taskMap).join(', ')}`) } } diff --git a/src/worker/ingestion/action-manager.ts b/src/worker/ingestion/action-manager.ts index df9d996d..5496fafa 100644 --- a/src/worker/ingestion/action-manager.ts +++ b/src/worker/ingestion/action-manager.ts @@ -1,6 +1,5 @@ -import { Action,PluginsServerConfig } from '../../types' +import { Action, PluginsServerConfig } from '../../types' import { DB } from '../../utils/db/db' -import { PubSub } from '../../utils/pubsub' import { status } from '../../utils/status' type ActionCache = Record @@ -8,67 +7,57 @@ type ActionCache = Record export class ActionManager { private ready: boolean private db: DB - public pubSub: PubSub private actionCache: ActionCache - constructor(db: DB, serverConfig: PluginsServerConfig) { + constructor(db: DB) { this.ready = false this.db = db - this.pubSub = new PubSub(serverConfig, { - 'fetch-action': async (message) => { - const actionId = parseInt(message) - const refetchedAction = await this.db.fetchAction(actionId) - if (refetchedAction) { - status.info( - 'đŸŋ', - actionId in this.actionCache - ? `Refetched action ID ${actionId} from DB` - : `Fetched new action ID ${actionId} from DB` - ) - this.actionCache[actionId] = refetchedAction - } else if (actionId in this.actionCache) { - status.info( - 'đŸŋ', - `Tried to fetch action ID ${actionId} from DB, but it wasn't found in DB, so deleted from cache instead` - ) - delete this.actionCache[actionId] - } else { - status.info( - 'đŸŋ', - `Tried to fetch action ID ${actionId} from DB, but it wasn't found in DB or cache, so did nothing instead` - ) - } - }, - 'delete-action': (message) => { - const actionId = parseInt(message) - if (actionId in this.actionCache) { - status.info('đŸŋ', `Deleted action ID ${actionId} from cache`) - delete this.actionCache[actionId] - } else { - status.info( - 'đŸŋ', - `Tried to delete action ID ${actionId} from cache, but it wasn't found in cache, so did nothing instead` - ) - } - }, - }) this.actionCache = {} } public async prepare(): Promise { this.actionCache = await this.db.fetchAllActionsMap() - await this.pubSub.start() this.ready = true } - public async close(): Promise { - await this.pubSub.stop() - } - public getAction(id: Action['id']): Action | undefined { if (!this.ready) { throw new Error('ActionManager is not ready! Run actionManager.prepare() before this') } return this.actionCache[id] } + + public async fetchAction(id: Action['id']): Promise { + const refetchedAction = await this.db.fetchAction(id) + if (refetchedAction) { + status.info( + 'đŸŋ', + id in this.actionCache ? `Refetched action ID ${id} from DB` : `Fetched new action ID ${id} from DB` + ) + this.actionCache[id] = refetchedAction + } else if (id in this.actionCache) { + status.info( + 'đŸŋ', + `Tried to fetch action ID ${id} from DB, but it wasn't found in DB, so deleted from cache instead` + ) + delete this.actionCache[id] + } else { + status.info( + 'đŸŋ', + `Tried to fetch action ID ${id} from DB, but it wasn't found in DB or cache, so did nothing instead` + ) + } + } + + public deleteAction(id: Action['id']): void { + if (id in this.actionCache) { + status.info('đŸŋ', `Deleted action ID ${id} from cache`) + delete this.actionCache[id] + } else { + status.info( + 'đŸŋ', + `Tried to delete action ID ${id} from cache, but it wasn't found in cache, so did nothing instead` + ) + } + } } diff --git a/src/worker/ingestion/process-event.ts b/src/worker/ingestion/process-event.ts index 3b2142f5..38ec3588 100644 --- a/src/worker/ingestion/process-event.ts +++ b/src/worker/ingestion/process-event.ts @@ -24,6 +24,7 @@ import { Client } from '../../utils/celery/client' import { DB } from '../../utils/db/db' import { KafkaProducerWrapper } from '../../utils/db/kafka-producer-wrapper' import { elementsToString, personInitialAndUTMProperties, sanitizeEventName, timeoutGuard } from '../../utils/db/utils' +import { PubSub } from '../../utils/pubsub' import { status } from '../../utils/status' import { castTimestampOrNow, filterIncrementProperties, UUID, UUIDT } from '../../utils/utils' import { ActionManager } from './action-manager' @@ -41,6 +42,7 @@ export class EventsProcessor { teamManager: TeamManager personManager: PersonManager actionManager: ActionManager + pubSub: PubSub constructor(pluginsServer: PluginsServer) { this.ready = false @@ -51,7 +53,11 @@ export class EventsProcessor { this.celery = new Client(pluginsServer.db, pluginsServer.CELERY_DEFAULT_QUEUE) this.teamManager = new TeamManager(pluginsServer.db) this.personManager = new PersonManager(pluginsServer) - this.actionManager = new ActionManager(pluginsServer.db, pluginsServer) + this.actionManager = new ActionManager(pluginsServer.db) + this.pubSub = new PubSub(pluginsServer, { + 'fetch-action': async (message) => await this.actionManager.fetchAction(parseInt(message)), + 'delete-action': (message) => this.actionManager.deleteAction(parseInt(message)), + }) this.posthog = nodePostHog('sTMFPsFhdP1Ssg', { fetch }) if (process.env.NODE_ENV === 'test') { @@ -61,11 +67,12 @@ export class EventsProcessor { public async prepare(): Promise { await this.actionManager.prepare() + await this.pubSub.start() this.ready = true } public async close(): Promise { - await this.actionManager.close() + await this.pubSub.stop() } public async processEvent( diff --git a/tests/worker/ingestion/action-manager.test.ts b/tests/worker/ingestion/action-manager.test.ts index 178eee31..ed0f2bad 100644 --- a/tests/worker/ingestion/action-manager.test.ts +++ b/tests/worker/ingestion/action-manager.test.ts @@ -12,11 +12,10 @@ describe('ActionManager()', () => { beforeEach(async () => { ;[server, closeServer] = await createServer() await resetTestDatabase() - actionManager = new ActionManager(server.db, server) + actionManager = new ActionManager(server.db) await actionManager.prepare() }) afterEach(async () => { - await actionManager.close() await closeServer() }) @@ -55,7 +54,7 @@ describe('ActionManager()', () => { ) // This is normally done by Django async in such a situation - await (actionManager.pubSub.taskMap['fetch-action']('67') as Promise) + await actionManager.fetchAction(67) const reloadedAction = actionManager.getAction(67) expect(reloadedAction).toMatchObject({ From 580ffc17b3bd88d2d138812b04e865fc2f75f3da Mon Sep 17 00:00:00 2001 From: Michael Matloka Date: Fri, 21 May 2021 04:17:08 +0200 Subject: [PATCH 03/21] Remove hello --- src/worker/tasks.ts | 3 --- 1 file changed, 3 deletions(-) diff --git a/src/worker/tasks.ts b/src/worker/tasks.ts index 487a2f2d..c5614e44 100644 --- a/src/worker/tasks.ts +++ b/src/worker/tasks.ts @@ -9,9 +9,6 @@ import { teardownPlugins } from './plugins/teardown' type TaskRunner = (server: PluginsServer, args: any) => Promise | any export const workerTasks: Record = { - hello: (server, args) => { - return `hello ${args}!` - }, onEvent: (server, args: { event: PluginEvent }) => { return runOnEvent(server, args.event) }, From 755e5dd3f570c63a41cb0b16c133a1dc874ed7dc Mon Sep 17 00:00:00 2001 From: Michael Matloka Date: Fri, 21 May 2021 11:59:46 +0200 Subject: [PATCH 04/21] Adjust ActionManager method names and use single PubSub --- benchmarks/postgres/ingestion.benchmark.ts | 1 - src/main/pluginsServer.ts | 6 +++++- src/worker/ingestion/action-manager.ts | 4 ++-- src/worker/ingestion/process-event.ts | 10 ---------- src/worker/tasks.ts | 8 +++++++- tests/shared/process-event.ts | 1 - tests/worker/ingestion/action-manager.test.ts | 2 +- 7 files changed, 15 insertions(+), 17 deletions(-) diff --git a/benchmarks/postgres/ingestion.benchmark.ts b/benchmarks/postgres/ingestion.benchmark.ts index bc6319b4..f3444dd7 100644 --- a/benchmarks/postgres/ingestion.benchmark.ts +++ b/benchmarks/postgres/ingestion.benchmark.ts @@ -63,7 +63,6 @@ describe('ingestion benchmarks', () => { }) afterEach(async () => { - await eventsProcessor.close() await stopServer?.() }) diff --git a/src/main/pluginsServer.ts b/src/main/pluginsServer.ts index 9c873c57..f56cf7fc 100644 --- a/src/main/pluginsServer.ts +++ b/src/main/pluginsServer.ts @@ -141,13 +141,17 @@ export async function startPluginsServer( void jobQueueConsumer?.resume() }) - // use one extra connection for Redis-based PubSub + // use one extra Redis connection for pub-sub pubSub = new PubSub(server, { [server.PLUGINS_RELOAD_PUBSUB_CHANNEL]: async () => { status.info('⚡', 'Reloading plugins!') await piscina?.broadcastTask({ task: 'reloadPlugins' }) await scheduleControl?.reloadSchedule() }, + 'reload-action': async (message) => + await piscina?.broadcastTask({ task: 'reloadAction', args: { actionId: parseInt(message) } }), + 'drop-action': async (message) => + await piscina?.broadcastTask({ task: 'dropAction`', args: { actionId: parseInt(message) } }), }) await pubSub.start() diff --git a/src/worker/ingestion/action-manager.ts b/src/worker/ingestion/action-manager.ts index 5496fafa..1acd2a45 100644 --- a/src/worker/ingestion/action-manager.ts +++ b/src/worker/ingestion/action-manager.ts @@ -27,7 +27,7 @@ export class ActionManager { return this.actionCache[id] } - public async fetchAction(id: Action['id']): Promise { + public async reloadAction(id: Action['id']): Promise { const refetchedAction = await this.db.fetchAction(id) if (refetchedAction) { status.info( @@ -49,7 +49,7 @@ export class ActionManager { } } - public deleteAction(id: Action['id']): void { + public dropAction(id: Action['id']): void { if (id in this.actionCache) { status.info('đŸŋ', `Deleted action ID ${id} from cache`) delete this.actionCache[id] diff --git a/src/worker/ingestion/process-event.ts b/src/worker/ingestion/process-event.ts index 38ec3588..15d8f467 100644 --- a/src/worker/ingestion/process-event.ts +++ b/src/worker/ingestion/process-event.ts @@ -42,7 +42,6 @@ export class EventsProcessor { teamManager: TeamManager personManager: PersonManager actionManager: ActionManager - pubSub: PubSub constructor(pluginsServer: PluginsServer) { this.ready = false @@ -54,10 +53,6 @@ export class EventsProcessor { this.teamManager = new TeamManager(pluginsServer.db) this.personManager = new PersonManager(pluginsServer) this.actionManager = new ActionManager(pluginsServer.db) - this.pubSub = new PubSub(pluginsServer, { - 'fetch-action': async (message) => await this.actionManager.fetchAction(parseInt(message)), - 'delete-action': (message) => this.actionManager.deleteAction(parseInt(message)), - }) this.posthog = nodePostHog('sTMFPsFhdP1Ssg', { fetch }) if (process.env.NODE_ENV === 'test') { @@ -67,14 +62,9 @@ export class EventsProcessor { public async prepare(): Promise { await this.actionManager.prepare() - await this.pubSub.start() this.ready = true } - public async close(): Promise { - await this.pubSub.stop() - } - public async processEvent( distinctId: string, ip: string | null, diff --git a/src/worker/tasks.ts b/src/worker/tasks.ts index c5614e44..21bca19e 100644 --- a/src/worker/tasks.ts +++ b/src/worker/tasks.ts @@ -1,6 +1,6 @@ import { PluginEvent } from '@posthog/plugin-scaffold/src/types' -import { EnqueuedJob, PluginsServer, PluginTaskType } from '../types' +import { Action, EnqueuedJob, PluginsServer, PluginTaskType } from '../types' import { ingestEvent } from './ingestion/ingest-event' import { runOnEvent, runOnSnapshot, runPluginTask, runProcessEvent, runProcessEventBatch } from './plugins/run' import { loadSchedule, setupPlugins } from './plugins/setup' @@ -45,6 +45,12 @@ export const workerTasks: Record = { reloadSchedule: async (server) => { await loadSchedule(server) }, + reloadAction: async (server, args: { actionId: Action['id'] }) => { + return await server.eventsProcessor.actionManager.reloadAction(args.actionId) + }, + dropAction: (server, args: { actionId: Action['id'] }) => { + return server.eventsProcessor.actionManager.dropAction(args.actionId) + }, teardownPlugins: async (server) => { await teardownPlugins(server) }, diff --git a/tests/shared/process-event.ts b/tests/shared/process-event.ts index d34aaf68..8a263ada 100644 --- a/tests/shared/process-event.ts +++ b/tests/shared/process-event.ts @@ -140,7 +140,6 @@ export const createProcessEventTests = ( }) afterEach(async () => { - await eventsProcessor.close() await server.redisPool.release(redis) await stopServer?.() }) diff --git a/tests/worker/ingestion/action-manager.test.ts b/tests/worker/ingestion/action-manager.test.ts index ed0f2bad..5823896d 100644 --- a/tests/worker/ingestion/action-manager.test.ts +++ b/tests/worker/ingestion/action-manager.test.ts @@ -54,7 +54,7 @@ describe('ActionManager()', () => { ) // This is normally done by Django async in such a situation - await actionManager.fetchAction(67) + await actionManager.reloadAction(67) const reloadedAction = actionManager.getAction(67) expect(reloadedAction).toMatchObject({ From 54f254ad70092ac5a209369a87fbd392a4b5d095 Mon Sep 17 00:00:00 2001 From: Michael Matloka Date: Fri, 21 May 2021 13:02:51 +0200 Subject: [PATCH 05/21] Touch tests up --- src/types.ts | 1 - src/utils/db/db.ts | 28 +---- src/utils/db/server.ts | 1 - tests/helpers/sql.ts | 1 - tests/postgres/teardown.test.ts | 2 +- tests/worker/ingestion/action-manager.test.ts | 115 +++++++++--------- 6 files changed, 61 insertions(+), 87 deletions(-) diff --git a/src/types.ts b/src/types.ts index 3edd5926..1ee7ccde 100644 --- a/src/types.ts +++ b/src/types.ts @@ -312,7 +312,6 @@ export interface RawOrganization { name: string created_at: string updated_at: string - available_features: string[] } /** Usable Team model. */ diff --git a/src/utils/db/db.ts b/src/utils/db/db.ts index 568457c9..365baaa5 100644 --- a/src/utils/db/db.ts +++ b/src/utils/db/db.ts @@ -704,20 +704,10 @@ export class DB { public async fetchAllActionsMap(): Promise> { const rawActions: RawAction[] = ( - await this.postgresQuery( - ` - SELECT * FROM posthog_action`, - undefined, - 'fetchActions' - ) + await this.postgresQuery(`SELECT * FROM posthog_action`, undefined, 'fetchActions') ).rows const actionSteps: ActionStep[] = ( - await this.postgresQuery( - ` - SELECT * FROM posthog_actionstep`, - undefined, - 'fetchActionSteps' - ) + await this.postgresQuery(`SELECT * FROM posthog_actionstep`, undefined, 'fetchActionSteps') ).rows const actionsMap: Record = {} for (const rawAction of rawActions) { @@ -731,23 +721,13 @@ export class DB { public async fetchAction(id: Action['id']): Promise { const rawActions: RawAction[] = ( - await this.postgresQuery( - ` - SELECT * FROM posthog_action WHERE id = $1`, - [id], - 'fetchActions' - ) + await this.postgresQuery(`SELECT * FROM posthog_action WHERE id = $1`, [id], 'fetchActions') ).rows if (!rawActions.length) { return null } const steps: ActionStep[] = ( - await this.postgresQuery( - ` - SELECT * FROM posthog_actionstep WHERE action_id = $1`, - [id], - 'fetchActionSteps' - ) + await this.postgresQuery(`SELECT * FROM posthog_actionstep WHERE action_id = $1`, [id], 'fetchActionSteps') ).rows const action: Action = { ...rawActions[0], steps } return action diff --git a/src/utils/db/server.ts b/src/utils/db/server.ts index 1dc4691b..e3da02f0 100644 --- a/src/utils/db/server.ts +++ b/src/utils/db/server.ts @@ -193,7 +193,6 @@ export async function createServer( clearInterval(eventLoopLagInterval) } server.mmdbUpdateJob?.cancel() - await server.eventsProcessor?.close() await server.jobQueueManager?.disconnectProducer() if (kafkaProducer) { clearInterval(kafkaProducer.flushInterval) diff --git a/tests/helpers/sql.ts b/tests/helpers/sql.ts index 1f360d57..77a17a38 100644 --- a/tests/helpers/sql.ts +++ b/tests/helpers/sql.ts @@ -122,7 +122,6 @@ export async function createUserTeamAndOrganization( personalization: '{}', setup_section_2_completed: true, for_internal_metrics: false, - available_features: [], } as RawOrganization) await insertRow(db, 'posthog_organizationmembership', { id: organizationMembershipId, diff --git a/tests/postgres/teardown.test.ts b/tests/postgres/teardown.test.ts index ce2b18c7..208c441e 100644 --- a/tests/postgres/teardown.test.ts +++ b/tests/postgres/teardown.test.ts @@ -82,7 +82,7 @@ describe('teardown', () => { expect(event1.properties.storage).toBe('nope') await piscina!.broadcastTask({ task: 'reloadPlugins' }) - await delay(3000) + await delay(2000) const event2 = await piscina!.runTask({ task: 'processEvent', args: { event: { ...defaultEvent } } }) expect(event2.properties.storage).toBe('tore down') diff --git a/tests/worker/ingestion/action-manager.test.ts b/tests/worker/ingestion/action-manager.test.ts index 5823896d..829b819e 100644 --- a/tests/worker/ingestion/action-manager.test.ts +++ b/tests/worker/ingestion/action-manager.test.ts @@ -1,10 +1,9 @@ import { PluginsServer, PropertyOperator } from '../../../src/types' import { createServer } from '../../../src/utils/db/server' -import { createRedis, delay } from '../../../src/utils/utils' import { ActionManager } from '../../../src/worker/ingestion/action-manager' import { resetTestDatabase } from '../../helpers/sql' -describe('ActionManager()', () => { +describe('ActionManager', () => { let server: PluginsServer let closeServer: () => Promise let actionManager: ActionManager @@ -19,67 +18,65 @@ describe('ActionManager()', () => { await closeServer() }) - describe('getAction()', () => { - it('returns the correct action', async () => { - const action = actionManager.getAction(67) + it('returns the correct action', async () => { + const action = actionManager.getAction(67) - expect(action).toMatchObject({ - id: 67, - name: 'Test Action', - deleted: false, - post_to_slack: false, - slack_message_format: '', - is_calculating: false, - steps: [ - { - id: 911, - action_id: 67, - tag_name: null, - text: null, - href: null, - selector: null, - url: null, - url_matching: null, - name: null, - event: null, - properties: [{ type: 'event', operator: PropertyOperator.Exact, key: 'foo', value: ['bar'] }], - }, - ], - }) + expect(action).toMatchObject({ + id: 67, + name: 'Test Action', + deleted: false, + post_to_slack: false, + slack_message_format: '', + is_calculating: false, + steps: [ + { + id: 911, + action_id: 67, + tag_name: null, + text: null, + href: null, + selector: null, + url: null, + url_matching: null, + name: null, + event: null, + properties: [{ type: 'event', operator: PropertyOperator.Exact, key: 'foo', value: ['bar'] }], + }, + ], + }) - await server.db.postgresQuery( - `UPDATE posthog_actionstep SET properties = jsonb_set(properties, '{0,key}', '"baz"') WHERE id = 911`, - undefined, - 'testKey' - ) + await server.db.postgresQuery( + `UPDATE posthog_actionstep SET properties = jsonb_set(properties, '{0,key}', '"baz"') WHERE id = 911`, + undefined, + 'testKey' + ) - // This is normally done by Django async in such a situation - await actionManager.reloadAction(67) - const reloadedAction = actionManager.getAction(67) + // This is normally dispatched by Django and broadcasted by Piscina + await actionManager.reloadAction(67) + const reloadedAction = actionManager.getAction(67) - expect(reloadedAction).toMatchObject({ - id: 67, - name: 'Test Action', - deleted: false, - post_to_slack: false, - slack_message_format: '', - is_calculating: false, - steps: [ - { - id: 911, - action_id: 67, - tag_name: null, - text: null, - href: null, - selector: null, - url: null, - url_matching: null, - name: null, - event: null, - properties: [{ type: 'event', operator: PropertyOperator.Exact, key: 'baz', value: ['bar'] }], - }, - ], - }) + expect(reloadedAction).toMatchObject({ + id: 67, + name: 'Test Action', + deleted: false, + post_to_slack: false, + slack_message_format: '', + is_calculating: false, + steps: [ + { + id: 911, + action_id: 67, + tag_name: null, + text: null, + href: null, + selector: null, + url: null, + url_matching: null, + name: null, + event: null, + properties: [{ type: 'event', operator: PropertyOperator.Exact, key: 'baz', value: ['bar'] }], + }, + ], }) }) }) From a87a0a0e0e4bd1fb26e4419d01e8b31f206e8c14 Mon Sep 17 00:00:00 2001 From: Michael Matloka Date: Mon, 24 May 2021 22:47:19 +0200 Subject: [PATCH 06/21] Make some adjustments --- src/types.ts | 1 + src/utils/db/db.ts | 4 +++- tests/helpers/sql.ts | 1 + 3 files changed, 5 insertions(+), 1 deletion(-) diff --git a/src/types.ts b/src/types.ts index 85d7f165..ce47a278 100644 --- a/src/types.ts +++ b/src/types.ts @@ -312,6 +312,7 @@ export interface RawOrganization { name: string created_at: string updated_at: string + available_features: string[] } /** Usable Team model. */ diff --git a/src/utils/db/db.ts b/src/utils/db/db.ts index 97a10096..1821e3d9 100644 --- a/src/utils/db/db.ts +++ b/src/utils/db/db.ts @@ -737,7 +737,9 @@ export class DB { actionsMap[rawAction.id] = { ...rawAction, steps: [] } } for (const actionStep of actionSteps) { - actionsMap[actionStep.action_id].steps.push(actionStep) + if (actionStep.action_id in actionsMap) { + actionsMap[actionStep.action_id].steps.push(actionStep) + } } return actionsMap } diff --git a/tests/helpers/sql.ts b/tests/helpers/sql.ts index 77a17a38..1f360d57 100644 --- a/tests/helpers/sql.ts +++ b/tests/helpers/sql.ts @@ -122,6 +122,7 @@ export async function createUserTeamAndOrganization( personalization: '{}', setup_section_2_completed: true, for_internal_metrics: false, + available_features: [], } as RawOrganization) await insertRow(db, 'posthog_organizationmembership', { id: organizationMembershipId, From ff0013e8c69dc2b22a21966299db8a18f69d796c Mon Sep 17 00:00:00 2001 From: Michael Matloka Date: Tue, 25 May 2021 00:21:43 +0200 Subject: [PATCH 07/21] Disable `status` stdout logs in test mode --- src/utils/status.ts | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/src/utils/status.ts b/src/utils/status.ts index 4db19cef..76ade2a2 100644 --- a/src/utils/status.ts +++ b/src/utils/status.ts @@ -1,5 +1,7 @@ import { threadId } from 'worker_threads' +import { determineNodeEnv, NodeEnv } from './utils' + export type StatusMethod = (icon: string, ...message: any[]) => void export interface StatusBlueprint { @@ -22,6 +24,9 @@ export class Status implements StatusBlueprint { } buildMethod(type: keyof StatusBlueprint): StatusMethod { + if (determineNodeEnv() == NodeEnv.Test) { + return () => {} // eslint-disable-line @typescript-eslint/no-empty-function + } return (icon: string, ...message: any[]) => { console[type](this.determinePrefix(), icon, ...message.filter(Boolean)) } From e12d648085804fda339cf4f99f7fb4e5650fcb5b Mon Sep 17 00:00:00 2001 From: Michael Matloka Date: Tue, 25 May 2021 00:36:14 +0200 Subject: [PATCH 08/21] Fix `status` --- src/utils/status.ts | 5 ++--- src/worker/vm/extensions/console.ts | 2 +- 2 files changed, 3 insertions(+), 4 deletions(-) diff --git a/src/utils/status.ts b/src/utils/status.ts index 76ade2a2..a1b589b9 100644 --- a/src/utils/status.ts +++ b/src/utils/status.ts @@ -1,7 +1,5 @@ import { threadId } from 'worker_threads' -import { determineNodeEnv, NodeEnv } from './utils' - export type StatusMethod = (icon: string, ...message: any[]) => void export interface StatusBlueprint { @@ -24,7 +22,8 @@ export class Status implements StatusBlueprint { } buildMethod(type: keyof StatusBlueprint): StatusMethod { - if (determineNodeEnv() == NodeEnv.Test) { + if (process.env.NODE_ENV?.toLowerCase() === 'test') { + // TODO: use determineNodeEnv() here return () => {} // eslint-disable-line @typescript-eslint/no-empty-function } return (icon: string, ...message: any[]) => { diff --git a/src/worker/vm/extensions/console.ts b/src/worker/vm/extensions/console.ts index c26ab0eb..4a92d13b 100644 --- a/src/worker/vm/extensions/console.ts +++ b/src/worker/vm/extensions/console.ts @@ -18,7 +18,7 @@ function consoleFormat(...args: unknown[]): string { export function createConsole(server: PluginsServer, pluginConfig: PluginConfig): ConsoleExtension { async function consolePersist(type: PluginLogEntryType, ...args: unknown[]): Promise { - if (determineNodeEnv() == NodeEnv.Development) { + if (determineNodeEnv() === NodeEnv.Development) { status.info('👉', `${type} in ${pluginDigest(pluginConfig.plugin!, pluginConfig.team_id)}:`, ...args) } From f3415c88525a31df8fc9be635778872640915bea Mon Sep 17 00:00:00 2001 From: Michael Matloka Date: Tue, 25 May 2021 01:15:08 +0200 Subject: [PATCH 09/21] Fix test problems --- tests/helpers/sql.ts | 6 +++--- tests/postgres/worker.test.ts | 4 ---- 2 files changed, 3 insertions(+), 7 deletions(-) diff --git a/tests/helpers/sql.ts b/tests/helpers/sql.ts index 1f360d57..f479279e 100644 --- a/tests/helpers/sql.ts +++ b/tests/helpers/sql.ts @@ -158,7 +158,7 @@ export async function createUserTeamAndOrganization( data_attributes: JSON.stringify(['data-attr']), }) await insertRow(db, 'posthog_action', { - id: 67, + id: teamId + 67, team_id: teamId, name: 'Test Action', created_at: new Date().toISOString(), @@ -171,8 +171,8 @@ export async function createUserTeamAndOrganization( last_calculated_at: new Date().toISOString(), } as RawAction) await insertRow(db, 'posthog_actionstep', { - id: 911, - action_id: 67, + id: teamId + 911, + action_id: teamId + 67, tag_name: null, text: null, href: null, diff --git a/tests/postgres/worker.test.ts b/tests/postgres/worker.test.ts index 6d9d2f92..14d9c9d0 100644 --- a/tests/postgres/worker.test.ts +++ b/tests/postgres/worker.test.ts @@ -236,10 +236,6 @@ describe('createTaskRunner()', () => { taskRunner = createTaskRunner(server) }) - it('handles `hello` task', async () => { - expect(await taskRunner({ task: 'hello', args: ['world'] })).toEqual('hello world!') - }) - it('handles `processEvent` task', async () => { mocked(runProcessEvent).mockReturnValue('runProcessEvent response' as any) From 3f5553a934cdc1d11d34d8afa93c50402d6e9aff Mon Sep 17 00:00:00 2001 From: Michael Matloka Date: Tue, 25 May 2021 14:11:37 +0200 Subject: [PATCH 10/21] Fix dropAction typo --- src/main/pluginsServer.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/main/pluginsServer.ts b/src/main/pluginsServer.ts index f56cf7fc..55174753 100644 --- a/src/main/pluginsServer.ts +++ b/src/main/pluginsServer.ts @@ -151,7 +151,7 @@ export async function startPluginsServer( 'reload-action': async (message) => await piscina?.broadcastTask({ task: 'reloadAction', args: { actionId: parseInt(message) } }), 'drop-action': async (message) => - await piscina?.broadcastTask({ task: 'dropAction`', args: { actionId: parseInt(message) } }), + await piscina?.broadcastTask({ task: 'dropAction', args: { actionId: parseInt(message) } }), }) await pubSub.start() From 63785f1872001ab5b552c1ea8ee1a2efd2485e4b Mon Sep 17 00:00:00 2001 From: Michael Matloka Date: Tue, 25 May 2021 14:12:02 +0200 Subject: [PATCH 11/21] Reload all ActionManager caches every 5 min --- src/main/pluginsServer.ts | 4 ++++ src/worker/ingestion/action-manager.ts | 7 ++++++- src/worker/tasks.ts | 3 +++ 3 files changed, 13 insertions(+), 1 deletion(-) diff --git a/src/main/pluginsServer.ts b/src/main/pluginsServer.ts index 55174753..d1babfa5 100644 --- a/src/main/pluginsServer.ts +++ b/src/main/pluginsServer.ts @@ -160,6 +160,10 @@ export async function startPluginsServer( await server!.db!.redisSet('@posthog-plugin-server/enabled-job-queues', queueString) } + // every 5 minutes all ActionManager caches are reloaded for eventual consistency + pingJob = schedule.scheduleJob('*/5 * * * *', async () => { + await piscina?.broadcastTask({ task: 'reloadAllActions' }) + }) // every 5 seconds set Redis keys @posthog-plugin-server/ping and @posthog-plugin-server/version pingJob = schedule.scheduleJob('*/5 * * * * *', async () => { await server!.db!.redisSet('@posthog-plugin-server/ping', new Date().toISOString(), 60, { diff --git a/src/worker/ingestion/action-manager.ts b/src/worker/ingestion/action-manager.ts index 1acd2a45..33659926 100644 --- a/src/worker/ingestion/action-manager.ts +++ b/src/worker/ingestion/action-manager.ts @@ -16,7 +16,7 @@ export class ActionManager { } public async prepare(): Promise { - this.actionCache = await this.db.fetchAllActionsMap() + await this.reloadAllActions() this.ready = true } @@ -27,6 +27,11 @@ export class ActionManager { return this.actionCache[id] } + public async reloadAllActions(): Promise { + this.actionCache = await this.db.fetchAllActionsMap() + status.info('đŸŋ', 'Fetched all actions from DB anew') + } + public async reloadAction(id: Action['id']): Promise { const refetchedAction = await this.db.fetchAction(id) if (refetchedAction) { diff --git a/src/worker/tasks.ts b/src/worker/tasks.ts index 21bca19e..6da5c5a3 100644 --- a/src/worker/tasks.ts +++ b/src/worker/tasks.ts @@ -45,6 +45,9 @@ export const workerTasks: Record = { reloadSchedule: async (server) => { await loadSchedule(server) }, + reloadAllActions: async (server) => { + return await server.eventsProcessor.actionManager.reloadAllActions() + }, reloadAction: async (server, args: { actionId: Action['id'] }) => { return await server.eventsProcessor.actionManager.reloadAction(args.actionId) }, From 1590d3cef827ddd75e1237a713a5ac95ad681180 Mon Sep 17 00:00:00 2001 From: Michael Matloka Date: Tue, 25 May 2021 15:53:55 +0200 Subject: [PATCH 12/21] Fix duplicate RawAction --- src/types.ts | 16 +--------------- 1 file changed, 1 insertion(+), 15 deletions(-) diff --git a/src/types.ts b/src/types.ts index 2977ecea..e16282f0 100644 --- a/src/types.ts +++ b/src/types.ts @@ -430,21 +430,6 @@ export interface CohortPeople { person_id: number } -/** Raw Action row from database. */ -export interface RawAction { - id: number - team_id: TeamId - name: string | null - created_at: string - created_by_id: number | null - deleted: boolean - post_to_slack: boolean - slack_message_format: string - is_calculating: boolean - updated_at: string - last_calculated_at: string -} - /** Sync with posthog/frontend/src/types.ts */ export enum PropertyOperator { Exact = 'exact', @@ -523,6 +508,7 @@ export interface ActionStep { /** Raw Action row from database. */ export interface RawAction { id: number + team_id: TeamId name: string | null created_at: string created_by_id: number | null From 7f0dc185dca3096ca5c8632715d814e68422b8cd Mon Sep 17 00:00:00 2001 From: Michael Matloka Date: Tue, 25 May 2021 15:54:22 +0200 Subject: [PATCH 13/21] Don't stringify JSONB column for `insertRow` --- tests/helpers/sql.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/helpers/sql.ts b/tests/helpers/sql.ts index ab9007f8..e4772d02 100644 --- a/tests/helpers/sql.ts +++ b/tests/helpers/sql.ts @@ -187,7 +187,7 @@ export async function createUserTeamAndOrganization( url_matching: null, name: null, event: null, - properties: JSON.stringify([{ type: 'event', operator: PropertyOperator.Exact, key: 'foo', value: ['bar'] }]), + properties: [{ type: 'event', operator: PropertyOperator.Exact, key: 'foo', value: ['bar'] }], }) } From d768f1148b6230c849d71f328f81623bcee0d01f Mon Sep 17 00:00:00 2001 From: Michael Matloka Date: Tue, 25 May 2021 16:12:08 +0200 Subject: [PATCH 14/21] It's a hub now --- tests/worker/ingestion/action-manager.test.ts | 74 +++++++++++++++++-- 1 file changed, 68 insertions(+), 6 deletions(-) diff --git a/tests/worker/ingestion/action-manager.test.ts b/tests/worker/ingestion/action-manager.test.ts index 829b819e..53ac9dc9 100644 --- a/tests/worker/ingestion/action-manager.test.ts +++ b/tests/worker/ingestion/action-manager.test.ts @@ -1,17 +1,17 @@ -import { PluginsServer, PropertyOperator } from '../../../src/types' -import { createServer } from '../../../src/utils/db/server' +import { Hub, PropertyOperator } from '../../../src/types' +import { createHub } from '../../../src/utils/db/hub' import { ActionManager } from '../../../src/worker/ingestion/action-manager' import { resetTestDatabase } from '../../helpers/sql' describe('ActionManager', () => { - let server: PluginsServer + let hub: Hub let closeServer: () => Promise let actionManager: ActionManager beforeEach(async () => { - ;[server, closeServer] = await createServer() + ;[hub, closeServer] = await createHub() await resetTestDatabase() - actionManager = new ActionManager(server.db) + actionManager = new ActionManager(hub.db) await actionManager.prepare() }) afterEach(async () => { @@ -45,7 +45,69 @@ describe('ActionManager', () => { ], }) - await server.db.postgresQuery( + await hub.db.postgresQuery( + `UPDATE posthog_actionstep SET properties = jsonb_set(properties, '{0,key}', '"baz"') WHERE id = 911`, + undefined, + 'testKey' + ) + + // This is normally dispatched by Django and broadcasted by Piscina + await actionManager.reloadAction(67) + const reloadedAction = actionManager.getAction(67) + + expect(reloadedAction).toMatchObject({ + id: 67, + name: 'Test Action', + deleted: false, + post_to_slack: false, + slack_message_format: '', + is_calculating: false, + steps: [ + { + id: 911, + action_id: 67, + tag_name: null, + text: null, + href: null, + selector: null, + url: null, + url_matching: null, + name: null, + event: null, + properties: [{ type: 'event', operator: PropertyOperator.Exact, key: 'baz', value: ['bar'] }], + }, + ], + }) + }) + + it('returns the correct action when reloaded via Piscina', async () => { + const action = actionManager.getAction(67) + + expect(action).toMatchObject({ + id: 67, + name: 'Test Action', + deleted: false, + post_to_slack: false, + slack_message_format: '', + is_calculating: false, + steps: [ + { + id: 911, + action_id: 67, + tag_name: null, + text: null, + href: null, + selector: null, + url: null, + url_matching: null, + name: null, + event: null, + properties: [{ type: 'event', operator: PropertyOperator.Exact, key: 'foo', value: ['bar'] }], + }, + ], + }) + + await hub.db.postgresQuery( `UPDATE posthog_actionstep SET properties = jsonb_set(properties, '{0,key}', '"baz"') WHERE id = 911`, undefined, 'testKey' From a5be2e43a9f2963ebe30d288280ad673e0972e69 Mon Sep 17 00:00:00 2001 From: Michael Matloka Date: Tue, 25 May 2021 16:12:30 +0200 Subject: [PATCH 15/21] Filter by Action.deleted --- src/utils/db/db.ts | 16 +++++++++++++--- 1 file changed, 13 insertions(+), 3 deletions(-) diff --git a/src/utils/db/db.ts b/src/utils/db/db.ts index 1821e3d9..1cf47cc3 100644 --- a/src/utils/db/db.ts +++ b/src/utils/db/db.ts @@ -727,10 +727,16 @@ export class DB { public async fetchAllActionsMap(): Promise> { const rawActions: RawAction[] = ( - await this.postgresQuery(`SELECT * FROM posthog_action`, undefined, 'fetchActions') + await this.postgresQuery(`SELECT * FROM posthog_action WHERE deleted = FALSE`, undefined, 'fetchActions') ).rows const actionSteps: ActionStep[] = ( - await this.postgresQuery(`SELECT * FROM posthog_actionstep`, undefined, 'fetchActionSteps') + await this.postgresQuery( + `SELECT posthog_actionstep.*, posthog_action.deleted FROM posthog_actionstep + JOIN posthog_action ON (posthog_action.id = posthog_actionstep.action_id) + WHERE posthog_action.deleted = FALSE`, + undefined, + 'fetchActionSteps' + ) ).rows const actionsMap: Record = {} for (const rawAction of rawActions) { @@ -746,7 +752,11 @@ export class DB { public async fetchAction(id: Action['id']): Promise { const rawActions: RawAction[] = ( - await this.postgresQuery(`SELECT * FROM posthog_action WHERE id = $1`, [id], 'fetchActions') + await this.postgresQuery( + `SELECT * FROM posthog_action WHERE id = $1 AND deleted = FALSE`, + [id], + 'fetchActions' + ) ).rows if (!rawActions.length) { return null From 5dde8df929bf5f1760824b7c92b071495bb922e3 Mon Sep 17 00:00:00 2001 From: Michael Matloka Date: Tue, 25 May 2021 18:01:38 +0200 Subject: [PATCH 16/21] Enhance ActionManager tests --- src/utils/db/hub.ts | 2 +- tests/worker/ingestion/action-manager.test.ts | 91 ++++--------------- 2 files changed, 21 insertions(+), 72 deletions(-) diff --git a/src/utils/db/hub.ts b/src/utils/db/hub.ts index 1dc8498e..53ea72b5 100644 --- a/src/utils/db/hub.ts +++ b/src/utils/db/hub.ts @@ -175,7 +175,7 @@ export async function createHub( // :TODO: This is only used on worker threads, not main hub.eventsProcessor = new EventsProcessor(hub as Hub) - await server.eventsProcessor.prepare() + await hub.eventsProcessor.prepare() hub.jobQueueManager = new JobQueueManager(hub as Hub) try { diff --git a/tests/worker/ingestion/action-manager.test.ts b/tests/worker/ingestion/action-manager.test.ts index 53ac9dc9..42b7f866 100644 --- a/tests/worker/ingestion/action-manager.test.ts +++ b/tests/worker/ingestion/action-manager.test.ts @@ -1,4 +1,4 @@ -import { Hub, PropertyOperator } from '../../../src/types' +import { Hub, PropertyOperator, RawAction } from '../../../src/types' import { createHub } from '../../../src/utils/db/hub' import { ActionManager } from '../../../src/worker/ingestion/action-manager' import { resetTestDatabase } from '../../helpers/sql' @@ -19,10 +19,13 @@ describe('ActionManager', () => { }) it('returns the correct action', async () => { - const action = actionManager.getAction(67) + const ACTION_ID = 69 + const ACTION_STEP_ID = 913 + + const action = actionManager.getAction(ACTION_ID) expect(action).toMatchObject({ - id: 67, + id: ACTION_ID, name: 'Test Action', deleted: false, post_to_slack: false, @@ -30,8 +33,8 @@ describe('ActionManager', () => { is_calculating: false, steps: [ { - id: 911, - action_id: 67, + id: ACTION_STEP_ID, + action_id: ACTION_ID, tag_name: null, text: null, href: null, @@ -46,17 +49,18 @@ describe('ActionManager', () => { }) await hub.db.postgresQuery( - `UPDATE posthog_actionstep SET properties = jsonb_set(properties, '{0,key}', '"baz"') WHERE id = 911`, - undefined, + `UPDATE posthog_actionstep SET properties = jsonb_set(properties, '{0,key}', '"baz"') WHERE id = $1`, + [ACTION_STEP_ID], 'testKey' ) // This is normally dispatched by Django and broadcasted by Piscina - await actionManager.reloadAction(67) - const reloadedAction = actionManager.getAction(67) + await actionManager.reloadAction(ACTION_ID) + + const reloadedAction = actionManager.getAction(ACTION_ID) expect(reloadedAction).toMatchObject({ - id: 67, + id: ACTION_ID, name: 'Test Action', deleted: false, post_to_slack: false, @@ -64,8 +68,8 @@ describe('ActionManager', () => { is_calculating: false, steps: [ { - id: 911, - action_id: 67, + id: ACTION_STEP_ID, + action_id: ACTION_ID, tag_name: null, text: null, href: null, @@ -78,67 +82,12 @@ describe('ActionManager', () => { }, ], }) - }) - - it('returns the correct action when reloaded via Piscina', async () => { - const action = actionManager.getAction(67) - - expect(action).toMatchObject({ - id: 67, - name: 'Test Action', - deleted: false, - post_to_slack: false, - slack_message_format: '', - is_calculating: false, - steps: [ - { - id: 911, - action_id: 67, - tag_name: null, - text: null, - href: null, - selector: null, - url: null, - url_matching: null, - name: null, - event: null, - properties: [{ type: 'event', operator: PropertyOperator.Exact, key: 'foo', value: ['bar'] }], - }, - ], - }) - - await hub.db.postgresQuery( - `UPDATE posthog_actionstep SET properties = jsonb_set(properties, '{0,key}', '"baz"') WHERE id = 911`, - undefined, - 'testKey' - ) // This is normally dispatched by Django and broadcasted by Piscina - await actionManager.reloadAction(67) - const reloadedAction = actionManager.getAction(67) + actionManager.dropAction(ACTION_ID) - expect(reloadedAction).toMatchObject({ - id: 67, - name: 'Test Action', - deleted: false, - post_to_slack: false, - slack_message_format: '', - is_calculating: false, - steps: [ - { - id: 911, - action_id: 67, - tag_name: null, - text: null, - href: null, - selector: null, - url: null, - url_matching: null, - name: null, - event: null, - properties: [{ type: 'event', operator: PropertyOperator.Exact, key: 'baz', value: ['bar'] }], - }, - ], - }) + const droppedAction = actionManager.getAction(ACTION_ID) + + expect(droppedAction).toBeUndefined() }) }) From 4af1ba71d50be5d38ced9b8eb819181af6e7c29d Mon Sep 17 00:00:00 2001 From: Michael Matloka Date: Tue, 25 May 2021 18:34:52 +0200 Subject: [PATCH 17/21] Add Action-syncing task runner tests --- src/worker/tasks.ts | 70 +++++++++++++++++------------------ tests/postgres/worker.test.ts | 39 ++++++++++++++++--- tests/schedule.test.ts | 4 +- 3 files changed, 70 insertions(+), 43 deletions(-) diff --git a/src/worker/tasks.ts b/src/worker/tasks.ts index 2c73b422..59f9ba78 100644 --- a/src/worker/tasks.ts +++ b/src/worker/tasks.ts @@ -6,58 +6,58 @@ import { runOnEvent, runOnSnapshot, runPluginTask, runProcessEvent, runProcessEv import { loadSchedule, setupPlugins } from './plugins/setup' import { teardownPlugins } from './plugins/teardown' -type TaskRunner = (server: Hub, args: any) => Promise | any +type TaskRunner = (hub: Hub, args: any) => Promise | any export const workerTasks: Record = { - onEvent: (server, args: { event: PluginEvent }) => { - return runOnEvent(server, args.event) + onEvent: (hub, args: { event: PluginEvent }) => { + return runOnEvent(hub, args.event) }, - onSnapshot: (server, args: { event: PluginEvent }) => { - return runOnSnapshot(server, args.event) + onSnapshot: (hub, args: { event: PluginEvent }) => { + return runOnSnapshot(hub, args.event) }, - processEvent: (server, args: { event: PluginEvent }) => { - return runProcessEvent(server, args.event) + processEvent: (hub, args: { event: PluginEvent }) => { + return runProcessEvent(hub, args.event) }, - processEventBatch: (server, args: { batch: PluginEvent[] }) => { - return runProcessEventBatch(server, args.batch) + processEventBatch: (hub, args: { batch: PluginEvent[] }) => { + return runProcessEventBatch(hub, args.batch) }, - runJob: (server, { job }: { job: EnqueuedJob }) => { - return runPluginTask(server, job.type, PluginTaskType.Job, job.pluginConfigId, job.payload) + runJob: (hub, { job }: { job: EnqueuedJob }) => { + return runPluginTask(hub, job.type, PluginTaskType.Job, job.pluginConfigId, job.payload) }, - runEveryMinute: (server, args: { pluginConfigId: number }) => { - return runPluginTask(server, 'runEveryMinute', PluginTaskType.Schedule, args.pluginConfigId) + runEveryMinute: (hub, args: { pluginConfigId: number }) => { + return runPluginTask(hub, 'runEveryMinute', PluginTaskType.Schedule, args.pluginConfigId) }, - runEveryHour: (server, args: { pluginConfigId: number }) => { - return runPluginTask(server, 'runEveryHour', PluginTaskType.Schedule, args.pluginConfigId) + runEveryHour: (hub, args: { pluginConfigId: number }) => { + return runPluginTask(hub, 'runEveryHour', PluginTaskType.Schedule, args.pluginConfigId) }, - runEveryDay: (server, args: { pluginConfigId: number }) => { - return runPluginTask(server, 'runEveryDay', PluginTaskType.Schedule, args.pluginConfigId) + runEveryDay: (hub, args: { pluginConfigId: number }) => { + return runPluginTask(hub, 'runEveryDay', PluginTaskType.Schedule, args.pluginConfigId) }, - getPluginSchedule: (server) => { - return server.pluginSchedule + getPluginSchedule: (hub) => { + return hub.pluginSchedule }, - ingestEvent: async (server, args: { event: PluginEvent }) => { - return await ingestEvent(server, args.event) + ingestEvent: async (hub, args: { event: PluginEvent }) => { + return await ingestEvent(hub, args.event) }, - reloadPlugins: async (server) => { - await setupPlugins(server) + reloadPlugins: async (hub) => { + await setupPlugins(hub) }, - reloadSchedule: async (server) => { - await loadSchedule(server) + reloadSchedule: async (hub) => { + await loadSchedule(hub) }, - reloadAllActions: async (server) => { - return await server.eventsProcessor.actionManager.reloadAllActions() + reloadAllActions: async (hub) => { + return await hub.eventsProcessor.actionManager.reloadAllActions() }, - reloadAction: async (server, args: { actionId: Action['id'] }) => { - return await server.eventsProcessor.actionManager.reloadAction(args.actionId) + reloadAction: async (hub, args: { actionId: Action['id'] }) => { + return await hub.eventsProcessor.actionManager.reloadAction(args.actionId) }, - dropAction: (server, args: { actionId: Action['id'] }) => { - return server.eventsProcessor.actionManager.dropAction(args.actionId) + dropAction: (hub, args: { actionId: Action['id'] }) => { + return hub.eventsProcessor.actionManager.dropAction(args.actionId) }, - teardownPlugins: async (server) => { - await teardownPlugins(server) + teardownPlugins: async (hub) => { + await teardownPlugins(hub) }, - flushKafkaMessages: async (server) => { - await server.kafkaProducer?.flush() + flushKafkaMessages: async (hub) => { + await hub.kafkaProducer?.flush() }, } diff --git a/tests/postgres/worker.test.ts b/tests/postgres/worker.test.ts index 2ed1ad56..2b612a25 100644 --- a/tests/postgres/worker.test.ts +++ b/tests/postgres/worker.test.ts @@ -4,9 +4,12 @@ import { mocked } from 'ts-jest/utils' import { ServerInstance, startPluginsServer } from '../../src/main/pluginsServer' import { loadPluginSchedule } from '../../src/main/services/schedule' -import { LogLevel } from '../../src/types' +import { Hub, LogLevel } from '../../src/types' import { Client } from '../../src/utils/celery/client' +import { createHub } from '../../src/utils/db/hub' +import { KafkaProducerWrapper } from '../../src/utils/db/kafka-producer-wrapper' import { delay, UUIDT } from '../../src/utils/utils' +import { ActionManager } from '../../src/worker/ingestion/action-manager' import { ingestEvent } from '../../src/worker/ingestion/ingest-event' import { makePiscina } from '../../src/worker/piscina' import { runPluginTask, runProcessEvent, runProcessEventBatch } from '../../src/worker/plugins/run' @@ -16,6 +19,7 @@ import { createTaskRunner } from '../../src/worker/worker' import { resetTestDatabase } from '../helpers/sql' import { setupPiscina } from '../helpers/worker' +jest.mock('../../src/worker/ingestion/action-manager') jest.mock('../../src/utils/db/sql') jest.mock('../../src/utils/status') jest.mock('../../src/worker/ingestion/ingest-event') @@ -38,6 +42,9 @@ function createEvent(index = 0): PluginEvent { beforeEach(() => { console.debug = jest.fn() + jest.spyOn(ActionManager.prototype, 'reloadAllActions') + jest.spyOn(ActionManager.prototype, 'reloadAction') + jest.spyOn(ActionManager.prototype, 'dropAction') }) test('piscina worker test', async () => { @@ -229,12 +236,16 @@ describe('queue logic', () => { describe('createTaskRunner()', () => { let taskRunner: any - let hub: any + let hub: Hub + let closeHub: () => Promise - beforeEach(() => { - hub = { mock: 'server' } + beforeEach(async () => { + ;[hub, closeHub] = await createHub() taskRunner = createTaskRunner(hub) }) + afterEach(async () => { + await closeHub() + }) it('handles `processEvent` task', async () => { mocked(runProcessEvent).mockReturnValue('runProcessEvent response' as any) @@ -302,6 +313,24 @@ describe('createTaskRunner()', () => { expect(loadSchedule).toHaveBeenCalled() }) + it('handles `reloadAllActions` task', async () => { + await taskRunner({ task: 'reloadAllActions' }) + + expect(hub.eventsProcessor.actionManager.reloadAllActions).toHaveBeenCalledWith() + }) + + it('handles `reloadAction` task', async () => { + await taskRunner({ task: 'reloadAction', args: { actionId: 777 } }) + + expect(hub.eventsProcessor.actionManager.reloadAction).toHaveBeenCalledWith(777) + }) + + it('handles `dropAction` task', async () => { + await taskRunner({ task: 'dropAction', args: { actionId: 777 } }) + + expect(hub.eventsProcessor.actionManager.dropAction).toHaveBeenCalledWith(777) + }) + it('handles `teardownPlugin` task', async () => { await taskRunner({ task: 'teardownPlugins' }) @@ -309,7 +338,7 @@ describe('createTaskRunner()', () => { }) it('handles `flushKafkaMessages` task', async () => { - hub.kafkaProducer = { flush: jest.fn() } + hub.kafkaProducer = ({ flush: jest.fn() } as unknown) as KafkaProducerWrapper await taskRunner({ task: 'flushKafkaMessages' }) diff --git a/tests/schedule.test.ts b/tests/schedule.test.ts index 13e1fa79..91675aca 100644 --- a/tests/schedule.test.ts +++ b/tests/schedule.test.ts @@ -118,9 +118,7 @@ describe('startSchedule', () => { ` await resetTestDatabase(testCode) piscina = setupPiscina(workerThreads, 10) - const [_hub, _closeHub] = await createHub({ LOG_LEVEL: LogLevel.Log, SCHEDULE_LOCK_TTL: 3 }) - hub = _hub - closeHub = _closeHub + ;[hub, closeHub] = await createHub({ LOG_LEVEL: LogLevel.Log, SCHEDULE_LOCK_TTL: 3 }) redis = await hub.redisPool.acquire() await redis.del(LOCKED_RESOURCE) From 830482784d3d51a094d39b154fa1f8c5bdd99175 Mon Sep 17 00:00:00 2001 From: Michael Matloka Date: Wed, 26 May 2021 12:04:49 +0200 Subject: [PATCH 18/21] Use `LOG_LEVEL=warn` in tests --- src/config/config.ts | 2 +- src/types.ts | 11 ++++++++++- src/utils/status.ts | 6 ++++-- 3 files changed, 15 insertions(+), 4 deletions(-) diff --git a/src/config/config.ts b/src/config/config.ts index 37743f10..043950c8 100644 --- a/src/config/config.ts +++ b/src/config/config.ts @@ -47,7 +47,7 @@ export function getDefaultConfig(): PluginsServerConfig { WORKER_CONCURRENCY: coreCount, TASK_TIMEOUT: 30, TASKS_PER_WORKER: 10, - LOG_LEVEL: LogLevel.Info, + LOG_LEVEL: isTestEnv ? LogLevel.Warn : LogLevel.Info, SENTRY_DSN: null, STATSD_HOST: null, STATSD_PORT: 8125, diff --git a/src/types.ts b/src/types.ts index e16282f0..48732879 100644 --- a/src/types.ts +++ b/src/types.ts @@ -16,12 +16,21 @@ import { EventsProcessor } from './worker/ingestion/process-event' import { LazyPluginVM } from './worker/vm/lazy' export enum LogLevel { + None = 'none', Debug = 'debug', Info = 'info', Log = 'log', Warn = 'warn', Error = 'error', - None = 'none', +} + +export const logLevelToNumber: Record = { + [LogLevel.None]: 0, + [LogLevel.Debug]: 10, + [LogLevel.Info]: 20, + [LogLevel.Log]: 30, + [LogLevel.Warn]: 40, + [LogLevel.Error]: 50, } export interface PluginsServerConfig extends Record { diff --git a/src/utils/status.ts b/src/utils/status.ts index a1b589b9..5fe9a66b 100644 --- a/src/utils/status.ts +++ b/src/utils/status.ts @@ -1,5 +1,8 @@ import { threadId } from 'worker_threads' +import { defaultConfig } from '../config/config' +import { logLevelToNumber } from '../types' + export type StatusMethod = (icon: string, ...message: any[]) => void export interface StatusBlueprint { @@ -22,8 +25,7 @@ export class Status implements StatusBlueprint { } buildMethod(type: keyof StatusBlueprint): StatusMethod { - if (process.env.NODE_ENV?.toLowerCase() === 'test') { - // TODO: use determineNodeEnv() here + if (logLevelToNumber[defaultConfig.LOG_LEVEL] < logLevelToNumber[type]) { return () => {} // eslint-disable-line @typescript-eslint/no-empty-function } return (icon: string, ...message: any[]) => { From c132a53994bb83982f1f1df0ce07c3c24f4dfb4a Mon Sep 17 00:00:00 2001 From: Michael Matloka Date: Wed, 26 May 2021 12:05:34 +0200 Subject: [PATCH 19/21] Don't `throw` error on unassociated channel pubsub --- src/utils/pubsub.ts | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/src/utils/pubsub.ts b/src/utils/pubsub.ts index 0884b07b..8f49dcc8 100644 --- a/src/utils/pubsub.ts +++ b/src/utils/pubsub.ts @@ -1,3 +1,4 @@ +import { captureException } from '@sentry/node' import { Redis } from 'ioredis' import { PluginsServerConfig } from '../types' @@ -31,10 +32,12 @@ export class PubSub { this.redis.on('message', (channel: string, message: string) => { const task: PubSubTask | undefined = this.taskMap[channel] if (!task) { - throw new Error( - `Received a pubsub message for unassociated channel ${channel}! Associated channels are: ${Object.keys( - this.taskMap - ).join(', ')}` + captureException( + new Error( + `Received a pubsub message for unassociated channel ${channel}! Associated channels are: ${Object.keys( + this.taskMap + ).join(', ')}` + ) ) } void task(message) From 408200381dfebf8ab36ecccdd146190ac0da33be Mon Sep 17 00:00:00 2001 From: Michael Matloka Date: Wed, 26 May 2021 12:39:10 +0200 Subject: [PATCH 20/21] Don't use defaultConfig in Status.buildMethod due to circular import --- src/utils/status.ts | 6 ------ 1 file changed, 6 deletions(-) diff --git a/src/utils/status.ts b/src/utils/status.ts index 5fe9a66b..4db19cef 100644 --- a/src/utils/status.ts +++ b/src/utils/status.ts @@ -1,8 +1,5 @@ import { threadId } from 'worker_threads' -import { defaultConfig } from '../config/config' -import { logLevelToNumber } from '../types' - export type StatusMethod = (icon: string, ...message: any[]) => void export interface StatusBlueprint { @@ -25,9 +22,6 @@ export class Status implements StatusBlueprint { } buildMethod(type: keyof StatusBlueprint): StatusMethod { - if (logLevelToNumber[defaultConfig.LOG_LEVEL] < logLevelToNumber[type]) { - return () => {} // eslint-disable-line @typescript-eslint/no-empty-function - } return (icon: string, ...message: any[]) => { console[type](this.determinePrefix(), icon, ...message.filter(Boolean)) } From 6212419ec9f561a9d96d3ec6e0c8982354ae670d Mon Sep 17 00:00:00 2001 From: Michael Matloka Date: Wed, 26 May 2021 13:03:17 +0200 Subject: [PATCH 21/21] Fix actions reload job var name --- src/main/pluginsServer.ts | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/main/pluginsServer.ts b/src/main/pluginsServer.ts index 416a77a8..3fd90256 100644 --- a/src/main/pluginsServer.ts +++ b/src/main/pluginsServer.ts @@ -44,6 +44,7 @@ export async function startPluginsServer( let pubSub: PubSub | undefined let hub: Hub | undefined let fastifyInstance: FastifyInstance | undefined + let actionsReloadJob: schedule.Job | undefined let pingJob: schedule.Job | undefined let statsJob: schedule.Job | undefined let piscina: Piscina | undefined @@ -74,6 +75,7 @@ export async function startPluginsServer( lastActivityCheck && clearInterval(lastActivityCheck) await queue?.stop() await pubSub?.stop() + actionsReloadJob && schedule.cancelJob(actionsReloadJob) pingJob && schedule.cancelJob(pingJob) statsJob && schedule.cancelJob(statsJob) await jobQueueConsumer?.stop() @@ -161,7 +163,7 @@ export async function startPluginsServer( } // every 5 minutes all ActionManager caches are reloaded for eventual consistency - pingJob = schedule.scheduleJob('*/5 * * * *', async () => { + actionsReloadJob = schedule.scheduleJob('*/5 * * * *', async () => { await piscina?.broadcastTask({ task: 'reloadAllActions' }) }) // every 5 seconds set Redis keys @posthog-plugin-server/ping and @posthog-plugin-server/version