Port Python process_event_ee over to plugin server

.eslintrc.js

@@ -1,7 +1,7 @@
 module.exports = {
     parser: '@typescript-eslint/parser',
     plugins: ['@typescript-eslint'],
-    extends: ['plugin:@typescript-eslint/recommended', 'prettier/@typescript-eslint'],
+    extends: ['plugin:@typescript-eslint/recommended', 'prettier', 'prettier/@typescript-eslint', 'prettier/react'],
     ignorePatterns: ['bin', 'dist', 'node_modules'],
     rules: {
         '@typescript-eslint/no-unused-vars': 'off',

.github/workflows/ci.yml

@@ -73,8 +73,15 @@ jobs:
               with:
                   node-version: 14
 
+            - uses: actions/cache@v2
+              with:
+                  path: ${{ env.pythonLocation }}
+                  key: ${{ env.pythonLocation }}-${{ hashFiles('posthog/requirements.txt') }}
+
             - name: Install requirements.txt dependencies with pip
-              run: python -m pip install --upgrade pip && python -m pip install -r posthog/requirements.txt
+              run: |
+                  pip install --upgrade pip
+                  pip install --upgrade --upgrade-strategy eager -r posthog/requirements.txt
 
             - name: Set up databases
               env:
@@ -93,6 +100,97 @@ jobs:
                   REDIS_URL: 'redis://localhost'
               run: cd plugin-server && yarn test:postgres
 
+    tests-clickhouse:
+        name: Tests / ClickHouse + Kafka
+        runs-on: ubuntu-20.04
+
+        services:
+            postgres:
+                image: postgres:12
+                env:
+                    POSTGRES_USER: postgres
+                    POSTGRES_PASSWORD: postgres
+                    POSTGRES_DB: test_posthog
+                ports: ['5432:5432']
+                options: --health-cmd pg_isready --health-interval 10s --health-timeout 5s --health-retries 5
+            redis:
+                image: redis
+                ports:
+                    - '6379:6379'
+                options: >-
+                    --health-cmd "redis-cli ping"
+                    --health-interval 10s
+                    --health-timeout 5s
+                    --health-retries 5
+
+        env:
+            REDIS_URL: 'redis://localhost'
+            CLICKHOUSE_HOST: 'localhost'
+            CLICKHOUSE_DATABASE: 'posthog_test'
+            KAFKA_ENABLED: 'true'
+            KAFKA_HOSTS: 'kafka:9092'
+
+        steps:
+            - name: Check out Django server for database setup
+              uses: actions/checkout@v2
+              with:
+                  repository: 'PostHog/posthog'
+                  path: 'posthog/'
+
+            - name: Check out plugin server
+              uses: actions/checkout@v2
+              with:
+                  path: 'plugin-server'
+
+            - name: Fix Kafka Hostname
+              run: |
+                  sudo bash -c 'echo "127.0.0.1 kafka zookeeper" >> /etc/hosts'
+                  ping -c 1 kafka
+                  ping -c 1 zookeeper
+
+            - name: Start Kafka, Clickhouse, Zookeeper
+              run: |
+                  cd posthog/ee
+                  docker-compose -f docker-compose.ch.yml up -d zookeeper kafka clickhouse
+
+            - name: Set up Python
+              uses: actions/setup-python@v2
+              with:
+                  python-version: 3.8
+
+            - name: Set up Node 14
+              uses: actions/setup-node@v2
+              with:
+                  node-version: 14
+
+            - uses: actions/cache@v2
+              with:
+                  path: ${{ env.pythonLocation }}
+                  key: ${{ env.pythonLocation }}-${{ hashFiles('posthog/requirements.txt') }}
+
+            - name: Install requirements.txt dependencies with pip
+              run: |
+                  pip install --upgrade pip
+                  pip install --upgrade --upgrade-strategy eager -r posthog/requirements.txt
+
+            - name: Set up databases
+              env:
+                  SECRET_KEY: 'abcdef' # unsafe - for testing only
+                  DATABASE_URL: 'postgres://postgres:postgres@localhost:5432/posthog'
+                  PRIMARY_DB: 'clickhouse'
+                  TEST: 'true'
+              run: python posthog/manage.py setup_test_environment
+
+            - name: Install package.json dependencies with Yarn
+              run: cd plugin-server && yarn
+
+            - name: Test with Jest
+              env:
+                  # Below DB name has `test_` prepended, as that's how Django (ran above) creates the test DB
+                  DATABASE_URL: 'postgres://postgres:postgres@localhost:5432/test_posthog'
+                  REDIS_URL: 'redis://localhost'
+              run: cd plugin-server && yarn test:clickhouse
+
     benchmarks-postgres:
         name: Benchmarks / Postgres + Redis
         runs-on: ubuntu-20.04
@@ -141,8 +239,15 @@ jobs:
               with:
                   node-version: 14
 
+            - uses: actions/cache@v2
+              with:
+                  path: ${{ env.pythonLocation }}
+                  key: ${{ env.pythonLocation }}-${{ hashFiles('posthog/requirements.txt') }}
+
             - name: Install requirements.txt dependencies with pip
-              run: python -m pip install --upgrade pip && python -m pip install -r posthog/requirements.txt
+              run: |
+                  pip install --upgrade pip
+                  pip install --upgrade --upgrade-strategy eager -r posthog/requirements.txt
 
             - name: Set up databases
               env:

.gitignore

@@ -5,3 +5,4 @@ yarn-error.log
 dist/
 .yalc/
 yalc.lock
+src/idl/protos.*

Dockerfile

@@ -2,9 +2,10 @@ FROM node:14
 WORKDIR /code
 
 COPY package.json yarn.lock .eslintrc.js .prettierrc ./
+COPY src/idl/ src/idl/
 RUN yarn install --frozen-lockfile
 
 COPY ./ ./
-RUN yarn compile
+RUN yarn compile:typescript
 
 CMD [ "node", "dist/src/index.js" ]

README.md

@@ -27,30 +27,38 @@ Let's get you developing the plugin server in no time:
 
 There's a multitude of settings you can use to control the plugin server. Use them as environment variables.
 
-| Name                          | Description                                                | Default value                         |
-| ----------------------------- | ---------------------------------------------------------- | ------------------------------------- |
-| DATABASE_URL                  | Postgres database URL                                      | `'postgres://localhost:5432/posthog'` |
-| REDIS_URL                     | Redis store URL                                            | `'redis://localhost'`                 |
-| BASE_DIR                      | base path for resolving local plugins                      | `'.'`                                 |
-| WORKER_CONCURRENCY            | number of concurrent worker threads                        | `0` – all cores                       |
-| TASKS_PER_WORKER              | number of parallel tasks per worker thread                 | `10`                                  |
-| SCHEDULE_LOCK_TTL             | How many seconds to hold the lock for the schedule         | `60`                                  |
-| CELERY_DEFAULT_QUEUE          | Celery outgoing queue                                      | `'celery'`                            |
-| PLUGINS_CELERY_QUEUE          | Celery incoming queue                                      | `'posthog-plugins'`                   |
-| PLUGINS_RELOAD_PUBSUB_CHANNEL | Redis channel for reload events                            | `'reload-plugins'`                    |
-| KAFKA_ENABLED                 | use Kafka instead of Celery to ingest events               | `false`                               |
-| KAFKA_HOSTS                   | comma-delimited Kafka hosts                                | `null`                                |
-| KAFKA_CLIENT_CERT_B64         | Kafka certificate in Base64                                | `null`                                |
-| KAFKA_CLIENT_CERT_KEY_B64     | Kafka certificate key in Base64                            | `null`                                |
-| KAFKA_TRUSTED_CERT_B64        | Kafka trusted CA in Base64                                 | `null`                                |
-| DISABLE_WEB                   | whether to disable web server                              | `true`                                |
-| WEB_PORT                      | port for web server to listen on                           | `3008`                                |
-| WEB_HOSTNAME                  | hostname for web server to listen on                       | `'0.0.0.0'`                           |
-| LOG_LEVEL                     | minimum log level                                          | `LogLevel.Info`                       |
-| SENTRY_DSN                    | Sentry ingestion URL                                       | `null`                                |
-| STATSD_HOST                   | StatsD host - integration disabled if this is not provided | `null`                                |
-| STATSD_PORT                   | StatsD port                                                | `8125`                                |
-| STATSD_PREFIX                 | StatsD prefix                                              | `'plugin-server.'`                    |
+| Name                          | Description                                                         | Default value                         |
+| ----------------------------- | ------------------------------------------------------------------- | ------------------------------------- |
+| DATABASE_URL                  | Postgres database URL                                               | `'postgres://localhost:5432/posthog'` |
+| REDIS_URL                     | Redis store URL                                                     | `'redis://localhost'`                 |
+| BASE_DIR                      | base path for resolving local plugins                               | `'.'`                                 |
+| WORKER_CONCURRENCY            | number of concurrent worker threads                                 | `0` – all cores                       |
+| TASKS_PER_WORKER              | number of parallel tasks per worker thread                          | `10`                                  |
+| SCHEDULE_LOCK_TTL             | How many seconds to hold the lock for the schedule                  | `60`                                  |
+| CELERY_DEFAULT_QUEUE          | Celery outgoing queue                                               | `'celery'`                            |
+| PLUGINS_CELERY_QUEUE          | Celery incoming queue                                               | `'posthog-plugins'`                   |
+| PLUGINS_RELOAD_PUBSUB_CHANNEL | Redis channel for reload events                                     | `'reload-plugins'`                    |
+| PLUGIN_SERVER_INGESTION       | Whether the plugin server should put events right into the database | `false`                               |
+| CLICKHOUSE_HOST               | ClickHouse host                                                     | `'localhost'`                         |
+| CLICKHOUSE_DATABASE           | ClickHouse database                                                 | `'default'`                           |
+| CLICKHOUSE_USER               | ClickHouse username                                                 | `'default'`                           |
+| CLICKHOUSE_PASSWORD           | ClickHouse password                                                 | `null`                                |
+| CLICKHOUSE_CA                 | ClickHouse CA certs                                                 | `null`                                |
+| CLICKHOUSE_SECURE             | Secure ClickHouse connection                                        | `false`                               |
+| KAFKA_ENABLED                 | use Kafka instead of Celery to ingest events                        | `false`                               |
+| KAFKA_HOSTS                   | comma-delimited Kafka hosts                                         | `null`                                |
+| KAFKA_CONSUMPTION_TOPIC       | Kafka consumption topic override                                    | `null` (automatic)                    |
+| KAFKA_CLIENT_CERT_B64         | Kafka certificate in Base64                                         | `null`                                |
+| KAFKA_CLIENT_CERT_KEY_B64     | Kafka certificate key in Base64                                     | `null`                                |
+| KAFKA_TRUSTED_CERT_B64        | Kafka trusted CA in Base64                                          | `null`                                |
+| DISABLE_WEB                   | whether to disable web server                                       | `true`                                |
+| WEB_PORT                      | port for web server to listen on                                    | `3008`                                |
+| WEB_HOSTNAME                  | hostname for web server to listen on                                | `'0.0.0.0'`                           |
+| LOG_LEVEL                     | minimum log level                                                   | `LogLevel.Info`                       |
+| SENTRY_DSN                    | Sentry ingestion URL                                                | `null`                                |
+| STATSD_HOST                   | StatsD host - integration disabled if this is not provided          | `null`                                |
+| STATSD_PORT                   | StatsD port                                                         | `8125`                                |
+| STATSD_PREFIX                 | StatsD prefix                                                       | `'plugin-server.'`                    |
 
 ## Questions?
 

jest.config.js

@@ -3,6 +3,6 @@ module.exports = {
     testEnvironment: 'node',
     clearMocks: true,
     coverageProvider: 'v8',
-    setupFilesAfterEnv: ['./jest.setup.kafka-mock.js', './jest.setup.fetch-mock.js'],
+    setupFilesAfterEnv: ['./jest.setup.fetch-mock.js'],
     testMatch: ['<rootDir>/tests/**/*.test.ts', '<rootDir>/benchmarks/**/*.benchmark.ts'],
 }

package.json

@@ -5,21 +5,29 @@
     "types": "dist/src/index.d.ts",
     "main": "dist/src/index.js",
     "scripts": {
-        "test": "jest --runInBand tests/*.test.ts",
-        "test:postgres": "yarn test --testPathIgnorePatterns '.*/clickhouse'",
-        "test:clickhouse": "yarn test --testPathIgnorePatterns '.*/postgres'",
+        "test": "jest --runInBand tests/**/*.test.ts",
+        "test:postgres": "jest --runInBand tests/postgres/*.test.ts tests/*.test.ts",
+        "test:clickhouse": "jest --runInBand tests/clickhouse/*.test.ts",
         "benchmark": "node --expose-gc node_modules/.bin/jest --runInBand benchmarks/",
         "start": "yarn start:dev",
         "start:dist": "node dist/src/index.js --base-dir ../posthog",
         "start:dev": "ts-node-dev --exit-child src/index.ts --base-dir ../posthog",
+        "start:dev:ee": "KAFKA_ENABLED=true KAFKA_HOSTS=localhost:9092 yarn start:dev",
         "build": "yarn clean && yarn compile",
         "clean": "rimraf dist/*",
-        "compile": "tsc -p .",
+        "compile:protobuf": "cd src/idl/ && rimraf protos.* && pbjs -t static-module -w commonjs -o protos.js *.proto && pbts -o protos.d.ts protos.js && eslint --fix . && prettier --write .",
+        "compile:typescript": "tsc",
+        "compile": "yarn compile:protobuf && yarn compile:typescript",
         "lint": "eslint .",
         "lint:fix": "eslint --fix .",
         "prettier": "prettier --write .",
         "prettier:check": "prettier --check .",
-        "prepublishOnly": "yarn build"
+        "prepare": "yarn compile:protobuf",
+        "prepublishOnly": "yarn build",
+        "setup:dev:clickhouse": "cd ../posthog && export DEBUG=1 PRIMARY_DB=clickhouse && source env/bin/activate && python manage.py migrate_clickhouse",
+        "setup:test:ee": "yarn setup:test:postgres && yarn setup:test:clickhouse",
+        "setup:test:postgres": "cd ../posthog && (dropdb test_posthog || echo 'no db to drop') && createdb test_posthog && source env/bin/activate && DATABASE_URL=postgres://localhost:5432/test_posthog DEBUG=1 python manage.py migrate",
+        "setup:test:clickhouse": "cd ../posthog && export TEST=1 PRIMARY_DB=clickhouse CLICKHOUSE_DATABASE=posthog_test && source env/bin/activate && python manage.py migrate_clickhouse"
     },
     "bin": {
         "posthog-plugin-server": "bin/posthog-plugin-server"
@@ -29,9 +37,9 @@
     "license": "MIT",
     "dependencies": {
         "@google-cloud/bigquery": "^5.5.0",
+        "@posthog/clickhouse": "^1.7.0",
         "@sentry/node": "^5.29.0",
         "@sentry/tracing": "^5.29.0",
-        "@types/luxon": "^1.25.0",
         "adm-zip": "^0.4.16",
         "fastify": "^3.8.0",
         "hot-shots": "^8.2.1",
@@ -42,6 +50,8 @@
         "node-schedule": "^1.3.2",
         "pg": "^8.4.2",
         "piscina": "^2.1.0",
+        "posthog-js-lite": "^0.0.5",
+        "protobufjs": "^6.10.2",
         "redlock": "^4.2.0",
         "tar-stream": "^2.1.4",
         "uuid": "^8.3.1",
@@ -53,10 +63,11 @@
         "@babel/core": "^7.0.0",
         "@babel/preset-env": "^7.0.0",
         "@babel/preset-typescript": "^7.8.3",
-        "@posthog/plugin-scaffold": "0.2.6",
+        "@posthog/plugin-scaffold": "0.2.8",
         "@types/adm-zip": "^0.4.33",
         "@types/ioredis": "^4.17.7",
         "@types/jest": "^26.0.15",
+        "@types/luxon": "^1.25.0",
         "@types/node": "^14.14.6",
         "@types/node-fetch": "^2.5.7",
         "@types/node-schedule": "^1.3.1",
@@ -95,6 +106,6 @@
     },
     "lint-staged": {
         "*.{js,ts}": "eslint --fix",
-        "*.{js,ts,css,scss,json,yml}": "prettier --write"
+        "*.{js,ts,css,scss,json,yml,md}": "prettier --write"
     }
 }

src/celery/worker.ts

@@ -171,14 +171,14 @@ export class Worker extends Base implements Queue {
                 throw new Error(`Missing process handler for task ${taskName}`)
             }
 
-            console.info(
+            console.debug(
                 `celery.node Received task: ${taskName}[${taskId}], args: ${args}, kwargs: ${JSON.stringify(kwargs)}`
             )
 
             const timeStart = process.hrtime()
             const taskPromise = handler(...args, kwargs).then((result) => {
                 const diff = process.hrtime(timeStart)
-                console.info(
+                console.debug(
                     `celery.node Task ${taskName}[${taskId}] succeeded in ${diff[0] + diff[1] / 1e9}s: ${result}`
                 )
                 this.activeTasks.delete(taskPromise)