feat: write recording summary events

[pauldambra]

Problem

see #15200 (comment)

When we store session recording events we materialize a lot of information using the snapshot data column.

We'll soon not be storing the snapshot data so won't be able to use that to materialize that information, so we need to capture it earlier in the pipeline. Since this is only used for searching for/summarizing recordings we don't need to store every event.

Changes

We'll push a summary event to a new kafka topic during ingestion. ClickHouse can ingest from that topic into an aggregating merge tree table. So that we store (in theory, although not in practice) only one row per session.

• add config to turn this on and off by team in plugin server
• add code behind that to write session recording summary events to a new topic in kafka
• add ClickHouse tables to ingest and aggregate those summary events

How did you test this code?

some developer tests and running it locally

[benjackwhite]

Whoop that was fast! Lots of changes needed but hopefully these are useful comments after the first draft.

[benjackwhite]

I never know if this is the case but isn't the missing ? after trim meaning that it could throw if there is no href?

Also why not use truthiness and keep it simple :D

Suggested change

	if (!!eventSummary.data.href?.trim().length && url === undefined) {
	if (!url && eventSummary.data.href?.trim()) {

[pauldambra]

The elvis operator says if the thing is there carry on if not return undefined

I prefer the explicit !! to remind myself I'm using truthiness but no strong opinion. I'll see what the tests think of the suggestion and go from there :)

[pauldambra]

Can investigate aggregations with

drop table if exists testing_things;

  CREATE TABLE IF NOT EXISTS testing_things
  (
      session_id VARCHAR,
      team_id Int64,
      distinct_id VARCHAR,
      timestamp DateTime64(6, 'UTC'),
      first_timestamp AggregateFunction(min, DateTime64(6, 'UTC')),
      last_timestamp AggregateFunction(max, DateTime64(6, 'UTC')),
      first_url Nullable(VARCHAR),
      click_count SimpleAggregateFunction(sum, Int64),
      keypress_count SimpleAggregateFunction(sum, Int64),
      mouse_activity_count SimpleAggregateFunction(sum, Int64)
  ) ENGINE = AggregatingMergeTree()
  PARTITION BY toYYYYMM(timestamp)
ORDER BY (team_id, session_id, toStartOfHour(timestamp));

insert into testing_things
select
    'wat',
    1,
    'wat',
    '2021-12-05T00:10:00',
    minState(toDateTime64('2021-12-05T00:10:00', 6, 'UTC')),
    maxState(toDateTime64('2021-12-05T00:10:00', 6, 'UTC')),
    anyState(null),
    4,
    8,
    12;

select session_id, team_id, any(distinct_id), minMerge(first_timestamp), maxMerge(last_timestamp), sum(click_count), sum(keypress_count), sum(mouse_activity_count)
from testing_things
group by session_id, team_id

[benjackwhite]

Looking better!

[pauldambra]

Hmmm, so close

[pauldambra]

I was using the same columns for kafka and the aggregating table because I'm a fool.

[benjackwhite]

To me this looks solid now 👍

[pauldambra]

@hazzadous I don't think the failing plugin server functional tests here are due to my change https://github.com/PostHog/posthog/actions/runs/4884989857/jobs/8719065927

Should I ignore?