feat: blobby ingestion warnings

frontend/src/scenes/data-management/ingestion-warnings/IngestionWarningsView.tsx

@@ -2,6 +2,8 @@ import { useValues } from 'kea'
 import { ReadingHog } from 'lib/components/hedgehogs'
 import { ProductIntroduction } from 'lib/components/ProductIntroduction/ProductIntroduction'
 import { TZLabel } from 'lib/components/TZLabel'
+import { IconPlayCircle } from 'lib/lemon-ui/icons'
+import { LemonButton } from 'lib/lemon-ui/LemonButton'
 import { LemonTable } from 'lib/lemon-ui/LemonTable'
 import { Link } from 'lib/lemon-ui/Link'
 import { Sparkline } from 'lib/lemon-ui/Sparkline'
@@ -19,6 +21,8 @@ const WARNING_TYPE_TO_DESCRIPTION = {
     event_timestamp_in_future: 'An event was sent more than 23 hours in the future',
     ingestion_capacity_overflow: 'Event ingestion has overflowed capacity',
     message_size_too_large: 'Discarded event exceeding 1MB limit',
+    replay_timestamp_invalid: 'Replay event timestamp is invalid',
+    replay_timestamp_too_far: 'Replay event timestamp was too far in the future',
 }
 
 const WARNING_TYPE_RENDERER = {
@@ -134,6 +138,68 @@ const WARNING_TYPE_RENDERER = {
             </>
         )
     },
+    replay_timestamp_invalid: function Render(warning: IngestionWarning): JSX.Element {
+        const details: {
+            timestamp: string
+            session_id: string
+        } = {
+            timestamp: warning.details.timestamp,
+            session_id: warning.details.replayRecord.session_id,
+        }
+        return (
+            <>
+                Session replay data dropped due to invalid timestamp::
+                <ul>
+                    <li>invalid timestamp: {details.timestamp}</li>
+                    <li>session_id: {details.session_id}</li>
+                </ul>
+                <div className="max-w-30 mt-2">
+                    <LemonButton
+                        type="primary"
+                        size="xsmall"
+                        to={urls.replaySingle(details.session_id)}
+                        sideIcon={<IconPlayCircle />}
+                        data-attr="skewed-timestamp-view-recording"
+                    >
+                        View recording
+                    </LemonButton>
+                </div>
+            </>
+        )
+    },
+    replay_timestamp_too_far: function Render(warning: IngestionWarning): JSX.Element {
+        const details: {
+            timestamp: string
+            session_id: string
+            daysFromNow: string
+        } = {
+            timestamp: warning.details.timestamp,
+            session_id: warning.details.replayRecord.session_id,
+            daysFromNow: warning.details.daysFromNow,
+        }
+        return (
+            <>
+                The session replay data timestamp was too different from the capture time, so the data was dropped.
+                Event values:
+                <ul>
+                    <li>invalid timestamp: {details.timestamp}</li>
+                    <li>session_id: {details.session_id}</li>
+                    <li>skew: {details.daysFromNow} days</li>
+                </ul>
+                <div className="max-w-30 mt-2">
+                    <LemonButton
+                        type="primary"
+                        size="xsmall"
+                        to={urls.replaySingle(details.session_id)}
+                        sideIcon={<IconPlayCircle />}
+                        data-attr="skewed-timestamp-view-recording"
+                    >
+                        View recording
+                    </LemonButton>
+                </div>
+            </>
+        )
+    },
 }
 
 export function IngestionWarningsView(): JSX.Element {

frontend/src/scenes/data-management/ingestion-warnings/__mocks__/ingestion-warnings-response.ts

@@ -2,6 +2,37 @@ import { dayjs } from 'lib/dayjs'
 
 export const ingestionWarningsResponse = (baseTime: dayjs.Dayjs): { results: Record<string, any> } => ({
     results: [
+        {
+            type: 'replay_timestamp_invalid',
+            lastSeen: baseTime.subtract(1, 'day'),
+            warnings: [
+                {
+                    type: 'replay_timestamp_invalid',
+                    timestamp: baseTime.subtract(1, 'day'),
+                    details: {
+                        timestamp: 'not a date',
+                        replayRecord: { session_id: 'some uuid' },
+                    },
+                },
+            ],
+            count: 1,
+        },
+        {
+            type: 'replay_timestamp_too_far',
+            lastSeen: baseTime.subtract(1, 'day'),
+            warnings: [
+                {
+                    type: 'replay_timestamp_too_far',
+                    timestamp: baseTime.subtract(1, 'day'),
+                    details: {
+                        timestamp: baseTime.add(4, 'day').toISOString(),
+                        replayRecord: { session_id: 'some uuid' },
+                        daysFromNow: 4,
+                    },
+                },
+            ],
+            count: 1,
+        },
         {
             type: 'event_timestamp_in_future',
             lastSeen: '2023-06-07T15:11:42.149000Z',

plugin-server/src/main/ingestion-queues/batch-processing/each-batch-ingestion.ts

@@ -145,9 +145,14 @@ export async function eachBatchParallelIngestion(
                     const distinct_id = currentBatch[0].pluginEvent.distinct_id
                     if (team && OverflowWarningLimiter.consume(`${team.id}:${distinct_id}`, 1)) {
                         processingPromises.push(
-                            captureIngestionWarning(queue.pluginsServer.db, team.id, 'ingestion_capacity_overflow', {
-                                overflowDistinctId: distinct_id,
-                            })
+                            captureIngestionWarning(
+                                queue.pluginsServer.db.kafkaProducer,
+                                team.id,
+                                'ingestion_capacity_overflow',
+                                {
+                                    overflowDistinctId: distinct_id,
+                                }
+                            )
                         )
                     }
                 }

plugin-server/src/main/ingestion-queues/session-recording/services/replay-events-ingester.ts

@@ -10,7 +10,9 @@ import { findOffsetsToCommit } from '../../../../kafka/consumer'
 import { retryOnDependencyUnavailableError } from '../../../../kafka/error-handling'
 import { createKafkaProducer, disconnectProducer, flushProducer, produce } from '../../../../kafka/producer'
 import { PluginsServerConfig } from '../../../../types'
+import { KafkaProducerWrapper } from '../../../../utils/db/kafka-producer-wrapper'
 import { status } from '../../../../utils/status'
+import { captureIngestionWarning } from '../../../../worker/ingestion/utils'
 import { eventDroppedCounter } from '../../metrics'
 import { createSessionReplayEvent } from '../process-event'
 import { IncomingRecordingMessage } from '../types'
@@ -130,7 +132,18 @@ export class ReplayEventsIngester {
                 // the replay record timestamp has to be valid and be within a reasonable diff from now
                 if (replayRecord !== null) {
                     const asDate = DateTime.fromSQL(replayRecord.first_timestamp)
-                    if (!asDate.isValid || Math.abs(asDate.diffNow('months').months) >= 0.99) {
+                    if (!asDate.isValid || Math.abs(asDate.diffNow('day').days) >= 7) {
+                        await captureIngestionWarning(
+                            new KafkaProducerWrapper(this.producer),
+                            event.team_id,
+                            !asDate.isValid ? 'replay_timestamp_invalid' : 'replay_timestamp_too_far',
+                            {
+                                replayRecord,
+                                timestamp: replayRecord.first_timestamp,
+                                isValid: asDate.isValid,
+                                daysFromNow: Math.round(Math.abs(asDate.diffNow('day').days)),
+                            }
+                        )
                         return drop('invalid_timestamp')
                     }
                 }

plugin-server/src/main/ingestion-queues/session-recording/session-recordings-consumer.ts

@@ -160,8 +160,8 @@ export class SessionRecordingIngester {
 
         this.realtimeManager = new RealtimeManager(this.redisPool, this.config)
 
-        // We create a hash of the cluster to use as a unique identifier for the high water marks
-        // This enables us to swap clusters without having to worry about resetting the high water marks
+        // We create a hash of the cluster to use as a unique identifier for the high-water marks
+        // This enables us to swap clusters without having to worry about resetting the high-water marks
         const kafkaClusterIdentifier = crypto.createHash('md5').update(this.config.KAFKA_HOSTS).digest('hex')
 
         this.sessionHighWaterMarker = new OffsetHighWaterMarker(

plugin-server/src/worker/ingestion/event-pipeline/populateTeamDataStep.ts

@@ -33,7 +33,7 @@ export async function populateTeamDataStep(
     if (event.team_id) {
         // Check for an invalid UUID, which should be blocked by capture, when team_id is present
         if (!UUID.validateString(event.uuid, false)) {
-            await captureIngestionWarning(db, event.team_id, 'skipping_event_invalid_uuid', {
+            await captureIngestionWarning(db.kafkaProducer, event.team_id, 'skipping_event_invalid_uuid', {
                 eventUuid: JSON.stringify(event.uuid),
             })
             throw new Error(`Not a valid UUID: "${event.uuid}"`)
@@ -70,7 +70,7 @@ export async function populateTeamDataStep(
 
     // Check for an invalid UUID, which should be blocked by capture, when team_id is present
     if (!UUID.validateString(event.uuid, false)) {
-        await captureIngestionWarning(db, team.id, 'skipping_event_invalid_uuid', {
+        await captureIngestionWarning(db.kafkaProducer, team.id, 'skipping_event_invalid_uuid', {
             eventUuid: JSON.stringify(event.uuid),
         })
         throw new Error(`Not a valid UUID: "${event.uuid}"`)

plugin-server/src/worker/ingestion/event-pipeline/prepareEventStep.ts

@@ -12,7 +12,7 @@ export async function prepareEventStep(runner: EventPipelineRunner, event: Plugi
     const invalidTimestampCallback = function (type: string, details: Record<string, any>) {
         invalidTimestampCounter.labels(type).inc()
 
-        tsParsingIngestionWarnings.push(captureIngestionWarning(runner.hub.db, team_id, type, details))
+        tsParsingIngestionWarnings.push(captureIngestionWarning(runner.hub.db.kafkaProducer, team_id, type, details))
     }
 
     const preIngestionEvent = await runner.hub.eventsProcessor.processEvent(

plugin-server/src/worker/ingestion/person-state.ts

@@ -323,15 +323,15 @@ export class PersonState {
             return undefined
         }
         if (isDistinctIdIllegal(mergeIntoDistinctId)) {
-            await captureIngestionWarning(this.db, teamId, 'cannot_merge_with_illegal_distinct_id', {
+            await captureIngestionWarning(this.db.kafkaProducer, teamId, 'cannot_merge_with_illegal_distinct_id', {
                 illegalDistinctId: mergeIntoDistinctId,
                 otherDistinctId: otherPersonDistinctId,
                 eventUuid: this.event.uuid,
             })
             return undefined
         }
         if (isDistinctIdIllegal(otherPersonDistinctId)) {
-            await captureIngestionWarning(this.db, teamId, 'cannot_merge_with_illegal_distinct_id', {
+            await captureIngestionWarning(this.db.kafkaProducer, teamId, 'cannot_merge_with_illegal_distinct_id', {
                 illegalDistinctId: otherPersonDistinctId,
                 otherDistinctId: mergeIntoDistinctId,
                 eventUuid: this.event.uuid,
@@ -404,7 +404,7 @@ export class PersonState {
         // If merge isn't allowed, we will ignore it, log an ingestion warning and exit
         if (!mergeAllowed) {
             // TODO: add event UUID to the ingestion warning
-            await captureIngestionWarning(this.db, this.teamId, 'cannot_merge_already_identified', {
+            await captureIngestionWarning(this.db.kafkaProducer, this.teamId, 'cannot_merge_already_identified', {
                 sourcePersonDistinctId: otherPersonDistinctId,
                 targetPersonDistinctId: mergeIntoDistinctId,
                 eventUuid: this.event.uuid,

plugin-server/src/worker/ingestion/process-event.ts

@@ -241,7 +241,7 @@ export class EventsProcessor {
                 // after plugin processing, person & group enrichment, etc.
                 if (error instanceof MessageSizeTooLarge) {
                     if (MessageSizeTooLargeWarningLimiter.consume(`${teamId}`, 1)) {
-                        await captureIngestionWarning(this.db, teamId, 'message_size_too_large', {
+                        await captureIngestionWarning(this.db.kafkaProducer, teamId, 'message_size_too_large', {
                             eventUuid: uuid,
                             distinctId: distinctId,
                         })

plugin-server/src/worker/ingestion/utils.ts

@@ -3,7 +3,7 @@ import { ProducerRecord } from 'kafkajs'
 import { DateTime } from 'luxon'
 
 import { PipelineEvent, TeamId, TimestampFormat } from '../../types'
-import { DB } from '../../utils/db/db'
+import { KafkaProducerWrapper } from '../../utils/db/kafka-producer-wrapper'
 import { safeClickhouseString } from '../../utils/db/utils'
 import { castTimestampOrNow, castTimestampToClickhouseFormat, UUIDT } from '../../utils/utils'
 import { KAFKA_EVENTS_DEAD_LETTER_QUEUE, KAFKA_INGESTION_WARNINGS } from './../../config/kafka-topics'
@@ -61,8 +61,13 @@ export function generateEventDeadLetterQueueMessage(
 // These get displayed under Data Management > Ingestion Warnings
 // These warnings get displayed to end users. Make sure these errors are actionable and useful for them and
 // also update IngestionWarningsView.tsx to display useful context.
-export async function captureIngestionWarning(db: DB, teamId: TeamId, type: string, details: Record<string, any>) {
-    await db.kafkaProducer.queueMessage({
+export async function captureIngestionWarning(
+    kafkaProducer: KafkaProducerWrapper,
+    teamId: TeamId,
+    type: string,
+    details: Record<string, any>
+) {
+    await kafkaProducer.queueMessage({
         topic: KAFKA_INGESTION_WARNINGS,
         messages: [
             {

plugin-server/tests/main/ingestion-queues/analytics-events-ingestion-overflow-consumer.test.ts

@@ -86,7 +86,7 @@ describe('eachBatchParallelIngestion with overflow consume', () => {
             expect(queue.pluginsServer.teamManager.getTeamForEvent).toHaveBeenCalledTimes(1)
             expect(consume).toHaveBeenCalledWith('1:id', 1)
             expect(captureIngestionWarning).toHaveBeenCalledWith(
-                queue.pluginsServer.db,
+                queue.pluginsServer.db.kafkaProducer,
                 1,
                 'ingestion_capacity_overflow',
                 {

plugin-server/tests/main/ingestion-queues/session-recording/services/replay-events-ingester.test.ts

@@ -1,12 +1,14 @@
+import { DateTime } from 'luxon'
 import { HighLevelProducer } from 'node-rdkafka'
 
 import { defaultConfig } from '../../../../../src/config/config'
 import { createKafkaProducer, produce } from '../../../../../src/kafka/producer'
 import { OffsetHighWaterMarker } from '../../../../../src/main/ingestion-queues/session-recording/services/offset-high-water-marker'
 import { ReplayEventsIngester } from '../../../../../src/main/ingestion-queues/session-recording/services/replay-events-ingester'
 import { IncomingRecordingMessage } from '../../../../../src/main/ingestion-queues/session-recording/types'
-import { PluginsServerConfig } from '../../../../../src/types'
+import { PluginsServerConfig, TimestampFormat } from '../../../../../src/types'
 import { status } from '../../../../../src/utils/status'
+import { castTimestampOrNow } from '../../../../../src/utils/utils'
 
 jest.mock('../../../../../src/utils/status')
 jest.mock('../../../../../src/kafka/producer')
@@ -47,6 +49,34 @@ describe('replay events ingester', () => {
         await ingester.start()
     })
 
+    test('does not ingest messages from a month in the future', async () => {
+        const twoMonthsFromNow = DateTime.utc().plus({ months: 2 })
+
+        await ingester.consume(makeIncomingMessage("mickey's fun house", twoMonthsFromNow.toMillis()))
+
+        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_ingestion_warnings_test')
+        // call.value is a Buffer convert it to a string
+        const value = call.value ? JSON.parse(call.value.toString()) : null
+        const expectedTimestamp = castTimestampOrNow(twoMonthsFromNow, TimestampFormat.ClickHouse)
+
+        expect(value.source).toEqual('plugin-server')
+        expect(value.team_id).toEqual(0)
+        expect(value.type).toEqual('replay_timestamp_too_far')
+        const details = JSON.parse(value.details)
+        expect(details).toEqual(
+            expect.objectContaining({
+                isValid: true,
+                daysFromNow: 61,
+                timestamp: expectedTimestamp,
+            })
+        )
+    })
+
     test('it passes snapshot source along', async () => {
         const ts = new Date().getTime()
         await ingester.consume(makeIncomingMessage("mickey's fun house", ts))
@@ -79,6 +109,7 @@ describe('replay events ingester', () => {
             uuid: expect.any(String),
         })
     })
+
     test('it defaults snapshot source to web when absent', async () => {
         const ts = new Date().getTime()
         await ingester.consume(makeIncomingMessage(null, ts))

plugin-server/tests/worker/ingestion/utils.test.ts

@@ -24,7 +24,7 @@ describe('captureIngestionWarning()', () => {
     }
 
     it('can read own writes', async () => {
-        await captureIngestionWarning(hub.db, 2, 'some_type', { foo: 'bar' })
+        await captureIngestionWarning(hub.db.kafkaProducer, 2, 'some_type', { foo: 'bar' })
 
         const warnings = await delayUntilEventIngested(fetchWarnings)
         expect(warnings).toEqual([