Process events in batches by mariusandra · Pull Request #39 · PostHog/plugin-server

mariusandra · 2020-12-07T09:39:33Z

~~Still WIP~~

Ready for review (comments below)

mariusandra · 2020-12-08T14:30:16Z

This is ready for a look!

Inside the PR:

Add support for a processEventBatch(batch: PluginEvent[], meta: PluginMeta) function inside plugins that can operate on a batch of events. For example to send 100 events at once to S3 instead of making 100 HTTP requests.
Either write processEvent or processEventBatch in your plugin. The other gets stubbed automatically.
Batches are split by team - all events in one batch are for the same team
Remove support for Posthog.capture inside processEvent and processEventBatch. Segment also doesn't support this in their destination functions. The problem is that since $identity events don't contain a token field, we can't initialize the posthog instance like we can for other events. We can add this back later by asking for the API key from postgres. However I'd just add it then to the scheduled/webhook plugins and not the processEvent side of thing to avoid infinite loops.
Update worker benchmark tests to see if batching makes a difference. (Spoiler: it does... will reply in the next comment)

mariusandra · 2020-12-08T14:45:31Z

I added one more fix since batching seemed to slow down sync events. Now in the stubbed processEventBatch function we only await Promise.all([...]) if we need to, greatly speeding things up!

Tested on a M1 Air. I got up to 100k events/sec with batches of 100. 🚀

// Before last fix
      ┌─────────┬────────────────┬───────────┬────────┬───────────┬──────────┬───────────┬───────────┬───────────┐
      │ (index) │    testName    │ coreCount │ events │ batchSize │ 1 thread │ 2 threads │ 4 threads │ 8 threads │
      ├─────────┼────────────────┼───────────┼────────┼───────────┼──────────┼───────────┼───────────┼───────────┤
      │    0    │    'simple'    │     8     │ 10000  │     1     │  39453   │   70847   │   81106   │   76728   │
      │    1    │    'simple'    │     8     │ 10000  │    10     │  24070   │   40898   │   54639   │   59051   │
      │    2    │    'simple'    │     8     │ 10000  │    100    │  30800   │   52167   │   69978   │   73826   │
      │    3    │   'for200k'    │     8     │ 10000  │     1     │  12650   │   22944   │   37368   │   43391   │
      │    4    │   'for200k'    │     8     │ 10000  │    10     │   9495   │   17370   │   27272   │   35226   │
      │    5    │   'for200k'    │     8     │ 10000  │    100    │  10329   │   19656   │   33916   │   37400   │
      │    6    │ 'timeout100ms' │     8     │ 10000  │     1     │   1010   │   1987    │   3352    │   5101    │
      │    7    │ 'timeout100ms' │     8     │ 10000  │    10     │   8839   │   16482   │   27245   │   31118   │
      │    8    │ 'timeout100ms' │     8     │ 10000  │    100    │  16397   │   29250   │   38102   │   38000   │
      └─────────┴────────────────┴───────────┴────────┴───────────┴──────────┴───────────┴───────────┴───────────┘

// After last fix
      ┌─────────┬────────────────┬───────────┬────────┬───────────┬──────────┬───────────┬───────────┬───────────┐
      │ (index) │    testName    │ coreCount │ events │ batchSize │ 1 thread │ 2 threads │ 4 threads │ 8 threads │
      ├─────────┼────────────────┼───────────┼────────┼───────────┼──────────┼───────────┼───────────┼───────────┤
      │    0    │    'simple'    │     8     │ 10000  │     1     │  44184   │   70350   │   79122   │   81577   │
      │    1    │    'simple'    │     8     │ 10000  │    10     │  31499   │   51685   │   75783   │   82023   │
      │    2    │    'simple'    │     8     │ 10000  │    100    │  46066   │   75966   │   99487   │  115690   │
      │    3    │   'for200k'    │     8     │ 10000  │     1     │  12684   │   23340   │   37095   │   42061   │
      │    4    │   'for200k'    │     8     │ 10000  │    10     │  10462   │   18992   │   31391   │   35820   │
      │    5    │   'for200k'    │     8     │ 10000  │    100    │  11667   │   22335   │   39604   │   43295   │
      │    6    │ 'timeout100ms' │     8     │ 10000  │     1     │   1011   │   1981    │   3417    │   5115    │
      │    7    │ 'timeout100ms' │     8     │ 10000  │    10     │   8801   │   16270   │   26732   │   31813   │
      │    8    │ 'timeout100ms' │     8     │ 10000  │    100    │  16475   │   28314   │   33264   │   38892   │
      └─────────┴────────────────┴───────────┴────────┴───────────┴──────────┴───────────┴───────────┴───────────┘

Twixes · 2020-12-09T07:57:35Z

+
+        for (const pluginConfig of pluginsToRun.reverse()) {
+            if (pluginConfig.vm?.methods?.processEventBatch && returnedEvents.length > 0) {
+                const { processEventBatch } = pluginConfig.vm.methods


Nit: could be a bit more concise with less lookups if we moved this line one level up (to right below the for (const pluginConfig... line) with optional chaining and in the if just checked for processEventBatch.

cleaned it up a bit

Twixes · 2020-12-09T07:58:24Z

+        // we have processEvent, but not processEventBatch
+        if (!__getExported('processEventBatch') && __getExported('processEvent')) {
+            exports.processEventBatch = async function processEventBatch (batch, meta) {
+                const processEvent = __getExported('processEvent');
+                let waitFor = false
+                const processedEvents = batch.map(event => {
+                    const e = processEvent(event, meta)
+                    if (e && typeof e.then !== 'undefined') {
+                        waitFor = true
+                    }
+                    return e
+                })
+                const response = waitFor ? (await Promise.all(processedEvents)) : processedEvents;
+                return response.filter(r => r)
+            }
+        // we have processEventBatch, but not processEvent


Where do we handle the case when there are both processEvent and processEventBatch?

The user handled it for us in that case. I think it's fine for the user to give both functions (e.g. making custom batching that only works with max 10 events), yet if they want to give just one of the two, the system should still work. Also if they give zero, the system should work... and then just do the plugin differently.

* silence some test logs * admit defeat in the battle with `esModuleInterop` * add basic typescript piscina workers * es module fix * remove rollup and compile directly with typescript * fix fetch import error * make piscina work in jest, dist and dev... run plugins through it * simplify tests * store plugin status on the `server` * config types * not all keys needed * add worker concurrency setting * clean up code * fix test worker code * add a basic benchmark test * refactor test * refactor test utils * support more serialized binary formats * mock jest behind enemy lines * less verbose logs in tests * add test to make sure adding more CPU cores makes plugins faster! * clone objects * add 12 * fix linter's worries * remove extra word * improve display * remove the exact cpu count from the list * add postgres and redis services to github to make tests pass * add ENV to config * add ENV to config * process events test wip * add db:init script * add db:init script to CI * add db:init script * add db:init script to CI * No prepublish on yarn install * processEvents in vm * run plugins on batches of events * tests for batching performance * async processEvent & events * lint * performance test batches * not a dev run * prettier for ts * prettier * remove support for PostHog.capture inside `processEvent` for now. This will go back inside the scheduled/webhook plugins * less round in tests, increase timeout * refactor * slight cleanup * rename `processEvents` to `processEventBatch` * only wait in processEventBatch if needed * Order file extensions more readably in lint-staged * Prettier * remove time rounding * less lookups * describe test devmode * add "light" * Remove redundant Math.round * Run prettier * Fix performance measurement * Prettier but correctly Co-authored-by: Michael Matloka <dev@twixes.com>

mariusandra added 30 commits December 3, 2020 14:11

silence some test logs

71acd57

admit defeat in the battle with esModuleInterop

c605b8d

add basic typescript piscina workers

cf0bc59

es module fix

cc8b50e

remove rollup and compile directly with typescript

6b31621

fix fetch import error

9185fdb

make piscina work in jest, dist and dev... run plugins through it

517ddb8

simplify tests

27fd502

store plugin status on the server

630cf58

config types

4456126

not all keys needed

2f0a872

add worker concurrency setting

6143d11

clean up code

2ed6fee

fix test worker code

92b2445

add a basic benchmark test

9b85afa

merge

c04f73d

refactor test

f4ec9a7

refactor test utils

45c90e9

support more serialized binary formats

4eec9e6

mock jest behind enemy lines

90b0913

less verbose logs in tests

8436771

add test to make sure adding more CPU cores makes plugins faster!

de14d97

clone objects

5d4685c

add 12

b96f373

fix linter's worries

e82f67f

remove extra word

6438aad

improve display

8f42315

remove the exact cpu count from the list

889cbaf

add postgres and redis services to github to make tests pass

ce933a7

add ENV to config

9886084

mariusandra added 4 commits December 8, 2020 14:55

less round in tests, increase timeout

29a89e3

refactor

9f65f77

slight cleanup

9f0a7ba

rename processEvents to processEventBatch

c4c8ac8

mariusandra requested a review from Twixes December 8, 2020 14:30

only wait in processEventBatch if needed

539a6d3

mariusandra mentioned this pull request Dec 8, 2020

Set right concurrency for plugin server on Heroku PostHog/posthog#2703

Merged

4 tasks

Twixes added 2 commits December 9, 2020 07:10

Merge branch 'master' into process-events

42dcee6

Order file extensions more readably in lint-staged

05976ba

Twixes reviewed Dec 9, 2020

View reviewed changes

Twixes and others added 5 commits December 9, 2020 09:15

Prettier

9193c54

remove time rounding

015e15f

less lookups

bdad14c

describe test devmode

92b8cbc

add "light"

0afd70e

mariusandra mentioned this pull request Dec 9, 2020

Refactor src/*.ts #47

Closed

Twixes added 2 commits December 9, 2020 10:14

Remove redundant Math.round

9dc93ac

Run prettier

0aaee9d

Twixes reviewed Dec 9, 2020

View reviewed changes

Comment thread src/plugins.ts Outdated

mariusandra commented Dec 9, 2020

View reviewed changes

Comment thread tests/worker.test.ts Outdated

Twixes added 2 commits December 9, 2020 10:25

Fix performance measurement

b43bce7

Prettier but correctly

fe28b45

mariusandra merged commit e2adfb4 into master Dec 9, 2020

mariusandra deleted the process-events branch December 9, 2020 10:09

This was referenced Dec 9, 2020

E2E benchmarks for celery and Kafka #56

Closed

Document Plugins for release v1.19.0 PostHog/posthog.com#757

Closed

Navigation Menu

Uh oh!

Process events in batches - #39

Process events in batches#39
mariusandra merged 70 commits into
masterfrom
process-events

mariusandra commented Dec 7, 2020 •

edited

Loading

Uh oh!

mariusandra commented Dec 8, 2020

Uh oh!

mariusandra commented Dec 8, 2020

Uh oh!

Uh oh!

Twixes Dec 9, 2020

Uh oh!

mariusandra Dec 9, 2020

Uh oh!

Twixes Dec 9, 2020

Uh oh!

mariusandra Dec 9, 2020

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Uh oh!

Conversation

mariusandra commented Dec 7, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

mariusandra commented Dec 8, 2020

Uh oh!

mariusandra commented Dec 8, 2020

Uh oh!

Uh oh!

Twixes Dec 9, 2020

Choose a reason for hiding this comment

Uh oh!

mariusandra Dec 9, 2020

Choose a reason for hiding this comment

Uh oh!

Twixes Dec 9, 2020

Choose a reason for hiding this comment

Uh oh!

mariusandra Dec 9, 2020

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

mariusandra commented Dec 7, 2020 •

edited

Loading