-
Notifications
You must be signed in to change notification settings - Fork 1.1k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
* export events via vm upgrade * cleaner exportEvents upgrade, add RetryError * add basic tests * test more things, fix buffer length issue * fix type * add missing vm method * add plugin scaffold to imports * Use JSDoc style for tooltips and fix onEvent typing * stringClamp * remove dead code * add consts * log locally * better log * it's a hub now * less events in benchmark to hopefully deflake a test * fix type bug * fix awkward bug * add statsd for export event jobs * typefix and rename * fix ! in test * flush on teardown * config as a string, as it should Co-authored-by: Michael Matloka <dev@twixes.com>
- Loading branch information
1 parent
a465836
commit a00dd0a
Showing
12 changed files
with
547 additions
and
56 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,152 @@ | ||
| import { createBuffer } from '@posthog/plugin-contrib' | ||
| import { Plugin, PluginEvent, PluginMeta, RetryError } from '@posthog/plugin-scaffold' | ||
|
|
||
| import { Hub, PluginConfig, PluginConfigVMInternalResponse, PluginTaskType } from '../../../types' | ||
| import { status } from '../../../utils/status' | ||
| import { stringClamp } from '../../../utils/utils' | ||
|
|
||
| const MAXIMUM_RETRIES = 15 | ||
| const EXPORT_BUFFER_BYTES_MINIMUM = 1 | ||
| const EXPORT_BUFFER_BYTES_DEFAULT = 1024 * 1024 | ||
| const EXPORT_BUFFER_BYTES_MAXIMUM = 100 * 1024 * 1024 | ||
| const EXPORT_BUFFER_SECONDS_MINIMUM = 1 | ||
| const EXPORT_BUFFER_SECONDS_DEFAULT = 10 | ||
| const EXPORT_BUFFER_SECONDS_MAXIMUM = 600 | ||
|
|
||
| type ExportEventsUpgrade = Plugin<{ | ||
| global: { | ||
| exportEventsBuffer: ReturnType<typeof createBuffer> | ||
| exportEventsToIgnore: Set<string> | ||
| exportEventsWithRetry: (payload: ExportEventsJobPayload, meta: PluginMeta<ExportEventsUpgrade>) => Promise<void> | ||
| } | ||
| config: { | ||
| exportEventsBufferBytes: string | ||
| exportEventsBufferSeconds: string | ||
| exportEventsToIgnore: string | ||
| } | ||
| jobs: { | ||
| exportEventsWithRetry: ExportEventsJobPayload | ||
| } | ||
| }> | ||
|
|
||
| interface ExportEventsJobPayload extends Record<string, any> { | ||
| batch: PluginEvent[] | ||
| batchId: number | ||
| retriesPerformedSoFar: number | ||
| } | ||
|
|
||
| /** | ||
| * Inject export abstraction code into plugin VM if it has method `exportEvents`: | ||
| * - add `global`/`config`/`jobs` stuff specified in the `ExportEventsUpgrade` type above, | ||
| * - patch `onEvent` with code to add the event to a buffer. | ||
| */ | ||
| export function upgradeExportEvents( | ||
| hub: Hub, | ||
| pluginConfig: PluginConfig, | ||
| response: PluginConfigVMInternalResponse<PluginMeta<ExportEventsUpgrade>> | ||
| ): void { | ||
| const { methods, tasks, meta } = response | ||
|
|
||
| if (!methods.exportEvents) { | ||
| return | ||
| } | ||
|
|
||
| const uploadBytes = stringClamp( | ||
| meta.config.exportEventsBufferBytes, | ||
| EXPORT_BUFFER_BYTES_DEFAULT, | ||
| EXPORT_BUFFER_BYTES_MINIMUM, | ||
| EXPORT_BUFFER_BYTES_MAXIMUM | ||
| ) | ||
| const uploadSeconds = stringClamp( | ||
| meta.config.exportEventsBufferSeconds, | ||
| EXPORT_BUFFER_SECONDS_DEFAULT, | ||
| EXPORT_BUFFER_SECONDS_MINIMUM, | ||
| EXPORT_BUFFER_SECONDS_MAXIMUM | ||
| ) | ||
|
|
||
| meta.global.exportEventsToIgnore = new Set( | ||
| meta.config.exportEventsToIgnore | ||
| ? meta.config.exportEventsToIgnore.split(',').map((event: string) => event.trim()) | ||
| : null | ||
| ) | ||
|
|
||
| meta.global.exportEventsBuffer = createBuffer({ | ||
| limit: uploadBytes, | ||
| timeoutSeconds: uploadSeconds, | ||
| onFlush: async (batch) => { | ||
| const jobPayload = { | ||
| batch, | ||
| batchId: Math.floor(Math.random() * 1000000), | ||
| retriesPerformedSoFar: 0, | ||
| } | ||
| // Running the first export code directly, without a job in between | ||
| await meta.global.exportEventsWithRetry(jobPayload, meta) | ||
| }, | ||
| }) | ||
|
|
||
| meta.global.exportEventsWithRetry = async ( | ||
| payload: ExportEventsJobPayload, | ||
| meta: PluginMeta<ExportEventsUpgrade> | ||
| ) => { | ||
| const start = new Date() | ||
| try { | ||
| await methods.exportEvents?.(payload.batch) | ||
| hub.statsd?.timing('plugin.export_events.success', start, { | ||
| plugin: pluginConfig.plugin?.name ?? '?', | ||
| teamId: pluginConfig.team_id.toString(), | ||
| }) | ||
| } catch (err) { | ||
| if (err instanceof RetryError) { | ||
| if (payload.retriesPerformedSoFar < MAXIMUM_RETRIES) { | ||
| const nextRetrySeconds = 2 ** (payload.retriesPerformedSoFar + 1) * 3 | ||
| await meta.jobs | ||
| .exportEventsWithRetry({ ...payload, retriesPerformedSoFar: payload.retriesPerformedSoFar + 1 }) | ||
| .runIn(nextRetrySeconds, 'seconds') | ||
|
|
||
| status.info( | ||
| '馃殐', | ||
| `Enqueued PluginConfig ${pluginConfig.id} batch ${payload.batchId} for retry #${ | ||
| payload.retriesPerformedSoFar + 1 | ||
| } in ${Math.round(nextRetrySeconds)}s` | ||
| ) | ||
| hub.statsd?.increment('plugin.export_events.retry_enqueued', { | ||
| retry: `${payload.retriesPerformedSoFar + 1}`, | ||
| plugin: pluginConfig.plugin?.name ?? '?', | ||
| teamId: pluginConfig.team_id.toString(), | ||
| }) | ||
| } else { | ||
| status.info( | ||
| '鈽狅笍', | ||
| `Dropped PluginConfig ${pluginConfig.id} batch ${payload.batchId} after retrying ${payload.retriesPerformedSoFar} times` | ||
| ) | ||
| hub.statsd?.increment('plugin.export_events.retry_dropped', { | ||
| retry: `${payload.retriesPerformedSoFar}`, | ||
| plugin: pluginConfig.plugin?.name ?? '?', | ||
| teamId: pluginConfig.team_id.toString(), | ||
| }) | ||
| } | ||
| } else { | ||
| throw err | ||
| } | ||
| } | ||
| } | ||
|
|
||
| tasks.job['exportEventsWithRetry'] = { | ||
| name: 'exportEventsWithRetry', | ||
| type: PluginTaskType.Job, | ||
| exec: (payload) => meta.global.exportEventsWithRetry(payload as ExportEventsJobPayload, meta), | ||
| } | ||
|
|
||
| const oldOnEvent = methods.onEvent | ||
| methods.onEvent = async (event: PluginEvent) => { | ||
| if (!meta.global.exportEventsToIgnore.has(event.event)) { | ||
| meta.global.exportEventsBuffer.add(event, JSON.stringify(event).length) | ||
| } | ||
| await oldOnEvent?.(event) | ||
| } | ||
|
|
||
| const oldTeardownPlugin = methods.teardownPlugin | ||
| methods.teardownPlugin = async () => { | ||
| await Promise.all([meta.global.exportEventsBuffer.flush(), oldTeardownPlugin?.()]) | ||
| } | ||
| } |
Oops, something went wrong.