Skip to content

Commit

Permalink
revert: "feat: add all urls to replay table" (#26784)
Browse files Browse the repository at this point in the history
  • Loading branch information
@pauldambra
pauldambra authored Dec 10, 2024
1 parent 020c37e commit 508697e
Show file tree
Hide file tree
Showing 8 changed files with 48 additions and 218 deletions.
22 changes: 12 additions & 10 deletions plugin-server/src/main/ingestion-queues/session-recording/process-event.ts
View file
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,6 @@ export interface SummarizedSessionRecordingEvent {
distinct_id: string
session_id: string
first_url: string | null
urls: string[]
click_count: number
keypress_count: number
mouse_activity_count: number
Expand Down Expand Up @@ -256,7 +255,7 @@ export const createSessionReplayEvent = (
session_id: string,
events: RRWebEvent[],
snapshot_source: string | null
): { event: SummarizedSessionRecordingEvent } => {
): { event: SummarizedSessionRecordingEvent; warnings: string[] } => {
const timestamps = getTimestampsFrom(events)

// but every event where chunk index = 0 must have an eventsSummary
Expand All @@ -269,14 +268,15 @@ export const createSessionReplayEvent = (
throw new Error('ignoring an empty session recording event')
}

const warnings: string[] = []

let clickCount = 0
let keypressCount = 0
let mouseActivity = 0
let consoleLogCount = 0
let consoleWarnCount = 0
let consoleErrorCount = 0
const urls: Set<string> = new Set()

let url: string | null = null
events.forEach((event) => {
if (event.type === RRWebEventType.IncrementalSnapshot) {
if (isClick(event)) {
Expand All @@ -291,8 +291,8 @@ export const createSessionReplayEvent = (
}

const eventUrl: string | undefined = hrefFrom(event)
if (eventUrl) {
urls.add(eventUrl)
if (url === null && eventUrl) {
url = eventUrl
}

if (event.type === RRWebEventType.Plugin && event.data?.plugin === 'rrweb/console@1') {
Expand All @@ -305,10 +305,13 @@ export const createSessionReplayEvent = (
consoleErrorCount += 1
}
}

if (event.type === RRWebEventType.Custom && event.data?.tag === 'Message too large') {
warnings.push('replay_message_too_large')
}
})

const activeTime = activeMilliseconds(events)
const urlArray = Array.from(urls)

// NB forces types to be correct e.g. by truncating or rounding
// to ensure we don't send floats when we should send an integer
Expand All @@ -322,8 +325,7 @@ export const createSessionReplayEvent = (
click_count: Math.trunc(clickCount),
keypress_count: Math.trunc(keypressCount),
mouse_activity_count: Math.trunc(mouseActivity),
first_url: urlArray.length ? urlArray[0] : null,
urls: urlArray,
first_url: url,
active_milliseconds: Math.round(activeTime),
console_log_count: Math.trunc(consoleLogCount),
console_warn_count: Math.trunc(consoleWarnCount),
Expand All @@ -334,5 +336,5 @@ export const createSessionReplayEvent = (
snapshot_source: snapshot_source || 'web',
}

return { event: data }
return { event: data, warnings }
}
22 changes: 19 additions & 3 deletions plugin-server/src/main/ingestion-queues/session-recording/services/replay-events-ingester.ts
View file
Original file line number Diff line number Diff line change
Expand Up @@ -71,9 +71,9 @@ export class ReplayEventsIngester {
status.error('🔁', '[replay-events] main_loop_error', { error })

if (error?.isRetriable) {
// We assume that if the error is retriable, then we
// We assume the if the error is retriable, then we
// are probably in a state where e.g. Kafka is down
// temporarily, and we would rather simply throw and
// temporarily and we would rather simply throw and
// have the process restarted.
throw error
}
Expand Down Expand Up @@ -124,7 +124,7 @@ export class ReplayEventsIngester {
try {
const rrwebEvents = Object.values(event.eventsByWindowId).reduce((acc, val) => acc.concat(val), [])

const { event: replayRecord } = createSessionReplayEvent(
const { event: replayRecord, warnings } = createSessionReplayEvent(
randomUUID(),
event.team_id,
event.distinct_id,
Expand Down Expand Up @@ -154,6 +154,22 @@ export class ReplayEventsIngester {
return drop('invalid_timestamp')
}
}

await Promise.allSettled(
warnings.map(async (warning) => {
await captureIngestionWarning(
new KafkaProducerWrapper(this.producer),
event.team_id,
warning,
{
replayRecord,
timestamp: replayRecord.first_timestamp,
processingTimestamp: DateTime.now().toISO(),
},
{ key: event.session_id }
)
})
)
} catch (e) {
captureException(e, {
extra: {
Expand Down
96 changes: 15 additions & 81 deletions plugin-server/tests/main/ingestion-queues/session-recording/process-event.test.ts
View file
Original file line number Diff line number Diff line change
Expand Up @@ -32,8 +32,8 @@ describe('session recording process event', () => {
| 'event_count'
| 'message_count'
| 'snapshot_source'
| 'urls'
>
expectedWarnings: string[]
}[] = [
{
testDescription: 'click and mouse counts are detected',
Expand Down Expand Up @@ -68,8 +68,8 @@ describe('session recording process event', () => {
event_count: 6,
message_count: 1,
snapshot_source: 'web',
urls: [],
},
expectedWarnings: [],
},
{
testDescription: 'keyboard press is detected',
Expand All @@ -92,8 +92,8 @@ describe('session recording process event', () => {
event_count: 1,
message_count: 1,
snapshot_source: 'web',
urls: [],
},
expectedWarnings: [],
},
{
testDescription: 'console log entries are counted',
Expand Down Expand Up @@ -181,8 +181,8 @@ describe('session recording process event', () => {
event_count: 11,
message_count: 1,
snapshot_source: 'web',
urls: [],
},
expectedWarnings: [],
},
{
testDescription: 'url can be detected in meta event',
Expand Down Expand Up @@ -219,8 +219,8 @@ describe('session recording process event', () => {
event_count: 2,
message_count: 1,
snapshot_source: 'web',
urls: ['http://127.0.0.1:8000/the/url'],
},
expectedWarnings: [],
},
{
testDescription: 'first url detection takes the first url whether meta url or payload url',
Expand Down Expand Up @@ -261,8 +261,8 @@ describe('session recording process event', () => {
event_count: 2,
message_count: 1,
snapshot_source: 'web',
urls: ['http://127.0.0.1:8000/home', 'http://127.0.0.1:8000/second/url'],
},
expectedWarnings: [],
},
{
testDescription: 'first url detection can use payload url',
Expand Down Expand Up @@ -307,8 +307,8 @@ describe('session recording process event', () => {
event_count: 2,
message_count: 1,
snapshot_source: 'web',
urls: ['http://127.0.0.1:8000/my-spa'],
},
expectedWarnings: [],
},
{
testDescription: 'negative timestamps are not included when picking timestamps',
Expand Down Expand Up @@ -336,8 +336,8 @@ describe('session recording process event', () => {
event_count: 3,
message_count: 1,
snapshot_source: 'web',
urls: [],
},
expectedWarnings: [],
},
{
testDescription: 'overlapping windows are summed separately for activity',
Expand Down Expand Up @@ -368,8 +368,8 @@ describe('session recording process event', () => {
event_count: 6,
message_count: 1,
snapshot_source: 'web',
urls: [],
},
expectedWarnings: [],
},
{
testDescription: 'mobile snapshot source is stored',
Expand All @@ -392,8 +392,8 @@ describe('session recording process event', () => {
mouse_activity_count: 1,
size: 82,
snapshot_source: 'mobile',
urls: [],
},
expectedWarnings: [],
},
{
testDescription: 'message too large warning is reported',
Expand All @@ -419,83 +419,15 @@ describe('session recording process event', () => {
mouse_activity_count: 1,
size: 169,
snapshot_source: 'web',
urls: [],
},
},
{
testDescription: 'urls array is deduplicated',
snapshotData: {
events_summary: [
{
timestamp: 1682449093469,
type: 5,
data: {
payload: {
// we don't read just any URL
'the-page-url': 'http://127.0.0.1:8000/not/included',
},
},
windowId: '1',
},
{
timestamp: 1682449093693,
type: 5,
data: {
payload: {
// matches href nested in payload
href: 'http://127.0.0.1:8000/my-spa',
},
},
windowId: '1',
},
{
timestamp: 1682449093693,
type: 5,
data: {
payload: {
// matches href nested in payload
href: 'http://127.0.0.1:8000/my-spa',
},
},
windowId: '1',
},
{
timestamp: 1682449093693,
type: 5,
data: {
payload: {
// matches href nested in payload
href: 'http://127.0.0.1:8000/my-spa/1',
},
},
windowId: '1',
},
],
},
expected: {
click_count: 0,
keypress_count: 0,
mouse_activity_count: 0,
first_url: 'http://127.0.0.1:8000/my-spa',
first_timestamp: '2023-04-25 18:58:13.469',
last_timestamp: '2023-04-25 18:58:13.693',
active_milliseconds: 0, // no data.source, so no activity
console_log_count: 0,
console_warn_count: 0,
console_error_count: 0,
size: 461,
event_count: 4,
message_count: 1,
snapshot_source: 'web',
urls: ['http://127.0.0.1:8000/my-spa', 'http://127.0.0.1:8000/my-spa/1'],
},
expectedWarnings: ['replay_message_too_large'],
},
]

it.each(sessionReplayEventTestCases)(
'session replay event generation - $testDescription',
({ snapshotData, snapshotSource, expected }) => {
const { event: data } = createSessionReplayEvent(
({ snapshotData, snapshotSource, expected, expectedWarnings }) => {
const { event: data, warnings } = createSessionReplayEvent(
'some-id',
12345,
'5AzhubH8uMghFHxXq0phfs14JOjH6SA2Ftr1dzXj7U4',
Expand All @@ -504,6 +436,8 @@ describe('session recording process event', () => {
snapshotSource || null
)

expect(warnings).toStrictEqual(expectedWarnings)

const expectedEvent: SummarizedSessionRecordingEvent = {
distinct_id: '5AzhubH8uMghFHxXq0phfs14JOjH6SA2Ftr1dzXj7U4',
session_id: 'abcf-efg',
Expand Down
56 changes: 2 additions & 54 deletions ...ver/tests/main/ingestion-queues/session-recording/services/replay-events-ingester.test.ts
View file
Original file line number Diff line number Diff line change
Expand Up @@ -12,18 +12,11 @@ import { castTimestampOrNow } from '../../../../../src/utils/utils'
jest.mock('../../../../../src/utils/status')
jest.mock('../../../../../src/kafka/producer')

const makeIncomingMessage = (
source: string | null,
timestamp: number,
extraWindowedEvents?: Record<string, Record<string, any>[]>
): IncomingRecordingMessage => {
const makeIncomingMessage = (source: string | null, timestamp: number): IncomingRecordingMessage => {
return {
distinct_id: '',
eventsRange: { start: timestamp, end: timestamp },
eventsByWindowId: {
'': [{ data: { any: 'thing' }, type: 2, timestamp: timestamp }],
...(extraWindowedEvents || {}),
},
eventsByWindowId: { '': [{ data: { any: 'thing' }, type: 2, timestamp: timestamp }] },
metadata: {
lowOffset: 0,
highOffset: 0,
Expand Down Expand Up @@ -112,7 +105,6 @@ describe('replay events ingester', () => {
snapshot_source: "mickey's fun house",
team_id: 0,
uuid: expect.any(String),
urls: [],
})
})

Expand Down Expand Up @@ -146,50 +138,6 @@ describe('replay events ingester', () => {
snapshot_source: 'web',
team_id: 0,
uuid: expect.any(String),
urls: [],
})
})

test('it adds URLs', async () => {
const ts = new Date().getTime()
await ingester.consume(
makeIncomingMessage("mickey's fun house", ts, {
anotherwindow: [
{ data: { href: 'thing' }, type: 2, timestamp: ts },
// should be deduplicated
{ data: { href: 'thing' }, type: 2, timestamp: ts },
{ data: { href: 'thing2' }, type: 2, timestamp: ts },
],
})
)

expect(jest.mocked(status.debug).mock.calls).toEqual([])
expect(jest.mocked(produce).mock.calls).toHaveLength(1)
expect(jest.mocked(produce).mock.calls[0]).toHaveLength(1)
const call = jest.mocked(produce).mock.calls[0][0]
expect(call.topic).toEqual('clickhouse_session_replay_events_test')
// call.value is a Buffer convert it to a string
const value = call.value ? JSON.parse(call.value.toString()) : null
expect(value).toEqual({
active_milliseconds: 0,
click_count: 0,
console_error_count: 0,
console_log_count: 0,
console_warn_count: 0,
distinct_id: '',
event_count: 4,
first_timestamp: expect.any(String),
first_url: 'thing',
keypress_count: 0,
last_timestamp: expect.any(String),
message_count: 1,
mouse_activity_count: 0,
session_id: '',
size: 245,
snapshot_source: "mickey's fun house",
team_id: 0,
uuid: expect.any(String),
urls: ['thing', 'thing2'],
})
})
})
Loading

0 comments on commit 508697e

Please sign in to comment.