feat: spike console logs

frontend/src/scenes/session-recordings/filters/AdvancedSessionRecordingsFilters.tsx

@@ -14,7 +14,7 @@ import {
 } from '~/types'
 import { DateFilter } from 'lib/components/DateFilter/DateFilter'
 import { DurationFilter } from './DurationFilter'
-import { LemonButtonWithDropdown, LemonCheckbox } from '@posthog/lemon-ui'
+import { LemonButtonWithDropdown, LemonCheckbox, LemonInput } from '@posthog/lemon-ui'
 import { TestAccountFilter } from 'scenes/insights/filters/TestAccountFilter'
 import { teamLogic } from 'scenes/teamLogic'
 import { useValues } from 'kea'
@@ -131,6 +131,17 @@ export const AdvancedSessionRecordingsFilters = ({
             <LemonLabel info="Show recordings that have captured console log messages">
                 Filter by console logs
             </LemonLabel>
+            <LemonInput
+                size={'small'}
+                fullWidth={true}
+                placeholder={'containing text'}
+                value={filters.console_search_query}
+                onChange={(s) => {
+                    setFilters({
+                        console_search_query: s,
+                    })
+                }}
+            />
             <ConsoleFilters
                 filters={filters}
                 setConsoleFilters={(x) =>

frontend/src/types.ts

@@ -749,6 +749,7 @@ export interface RecordingFilters {
     properties?: AnyPropertyFilter[]
     session_recording_duration?: RecordingDurationFilter
     duration_type_filter?: DurationType
+    console_search_query?: string
     console_logs?: FilterableLogLevel[]
     filter_test_accounts?: boolean
 }

plugin-server/src/config/kafka-topics.ts

@@ -37,3 +37,6 @@ export const KAFKA_CLICKHOUSE_SESSION_RECORDING_EVENTS = `${prefix}clickhouse_se
 export const KAFKA_CLICKHOUSE_SESSION_REPLAY_EVENTS = `${prefix}clickhouse_session_replay_events${suffix}`
 // write performance events to ClickHouse
 export const KAFKA_PERFORMANCE_EVENTS = `${prefix}clickhouse_performance_events${suffix}`
+
+// log entries for ingestion into clickhouse
+export const KAFKA_LOG_ENTRIES = `${prefix}log_entries${suffix}`

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

@@ -4,7 +4,7 @@ import { DateTime } from 'luxon'
 import { HighLevelProducer as RdKafkaProducer, NumberNullUndefined } from 'node-rdkafka-acosom'
 import { Counter } from 'prom-client'
 
-import { KAFKA_CLICKHOUSE_SESSION_REPLAY_EVENTS } from '../../../../config/kafka-topics'
+import { KAFKA_CLICKHOUSE_SESSION_REPLAY_EVENTS, KAFKA_LOG_ENTRIES } from '../../../../config/kafka-topics'
 import { createRdConnectionConfigFromEnvVars } from '../../../../kafka/config'
 import { findOffsetsToCommit } from '../../../../kafka/consumer'
 import { retryOnDependencyUnavailableError } from '../../../../kafka/error-handling'
@@ -116,7 +116,7 @@ export class ReplayEventsIngester {
         }
 
         try {
-            const replayRecord = createSessionReplayEvent(
+            const [replayRecord, consoleLogEntries] = createSessionReplayEvent(
                 randomUUID(),
                 event.team_id,
                 event.distinct_id,
@@ -160,13 +160,25 @@ export class ReplayEventsIngester {
 
             replayEventsCounter.inc()
 
+            const s1 = JSON.stringify(replayRecord)
+            const consoleLogProduces = consoleLogEntries.map((consoleLogEntry) => {
+                const s = JSON.stringify(consoleLogEntry)
+                status.info('🔁', '[replay-events] console_log', { s, s1 })
+                return produce({
+                    producer: this.producer,
+                    topic: KAFKA_LOG_ENTRIES,
+                    value: Buffer.from(s),
+                    key: event.session_id,
+                })
+            })
             return [
                 produce({
                     producer: this.producer,
                     topic: KAFKA_CLICKHOUSE_SESSION_REPLAY_EVENTS,
-                    value: Buffer.from(JSON.stringify(replayRecord)),
+                    value: Buffer.from(s1),
                     key: event.session_id,
                 }),
+                ...consoleLogProduces,
             ]
         } catch (error) {
             status.error('⚠️', '[replay-events] processing_error', {

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

@@ -5,6 +5,7 @@ import { DateTime } from 'luxon'
 
 import { activeMilliseconds } from '../../main/ingestion-queues/session-recording/snapshot-segmenter'
 import {
+    ClickHouseTimestamp,
     Element,
     GroupTypeIndex,
     Hub,
@@ -286,13 +287,39 @@ export interface SummarizedSessionRecordingEvent {
     message_count: number
 }
 
+type ConsoleLogEntry = {
+    team_id: number
+    message: string
+    log_level: 'info' | 'warn' | 'error'
+    log_source: 'session_replay'
+    // the session_id
+    log_source_id: string
+    // maybe we omit this?
+    instance_id: string
+    timestamp: ClickHouseTimestamp
+}
+
+function safeString(payload: string[]) {
+    let candidate = payload.join(' ')
+
+    if (candidate.startsWith('"') || candidate.startsWith("'")) {
+        candidate = candidate.substring(1)
+    }
+
+    if (candidate.endsWith('"') || candidate.endsWith("'")) {
+        candidate = candidate.substring(0, candidate.length - 1)
+    }
+
+    return candidate
+}
+
 export const createSessionReplayEvent = (
     uuid: string,
     team_id: number,
     distinct_id: string,
     session_id: string,
     events: RRWebEvent[]
-) => {
+): [SummarizedSessionRecordingEvent, ConsoleLogEntry[]] => {
     const timestamps = events
         .filter((e) => !!e?.timestamp)
         .map((e) => castTimestampOrNow(DateTime.fromMillis(e.timestamp), TimestampFormat.ClickHouse))
@@ -315,6 +342,8 @@ export const createSessionReplayEvent = (
     let consoleWarnCount = 0
     let consoleErrorCount = 0
     let url: string | null = null
+    const consoleLogEntries: ConsoleLogEntry[] = []
+
     events.forEach((event) => {
         if (event.type === 3) {
             mouseActivity += 1
@@ -337,6 +366,16 @@ export const createSessionReplayEvent = (
             } else if (level === 'error') {
                 consoleErrorCount += 1
             }
+            consoleLogEntries.push({
+                team_id,
+                // when is it not a single item array?
+                message: safeString(event.data.payload?.payload),
+                log_level: level,
+                log_source: 'session_replay',
+                log_source_id: session_id,
+                instance_id: event.data.instanceId,
+                timestamp: castTimestampOrNow(DateTime.fromMillis(event.timestamp), TimestampFormat.ClickHouse),
+            })
         }
     })
 
@@ -364,7 +403,7 @@ export const createSessionReplayEvent = (
         message_count: 1,
     }
 
-    return data
+    return [data, consoleLogEntries]
 }
 
 export function createPerformanceEvent(uuid: string, team_id: number, distinct_id: string, properties: Properties) {

posthog/api/__init__.py

@@ -4,6 +4,8 @@
 from posthog.batch_exports import http as batch_exports
 from posthog.settings import EE_AVAILABLE
 from posthog.warehouse.api import saved_query, table, view_link
+
+from ..session_recordings.session_recording_api import SessionRecordingViewSet
 from . import (
     activity_log,
     annotation,
@@ -40,7 +42,6 @@
 )
 from .dashboards import dashboard, dashboard_templates
 from .data_management import DataManagementViewSet
-from ..session_recordings.session_recording_api import SessionRecordingViewSet
 
 
 @decorators.api_view(["GET", "HEAD", "POST", "PUT", "PATCH", "DELETE"])
@@ -132,7 +133,22 @@ def api_not_found(request):
 batch_exports_router = projects_router.register(
     r"batch_exports", batch_exports.BatchExportViewSet, "batch_exports", ["team_id"]
 )
-batch_exports_router.register(r"runs", batch_exports.BatchExportRunViewSet, "runs", ["team_id", "batch_export_id"])
+batch_export_runs_router = batch_exports_router.register(
+    r"runs", batch_exports.BatchExportRunViewSet, "runs", ["team_id", "batch_export_id"]
+)
+batch_exports_router.register(
+    r"logs",
+    batch_exports.BatchExportLogViewSet,
+    "batch_export_run_logs",
+    ["team_id", "batch_export_id"],
+)
+
+batch_export_runs_router.register(
+    r"logs",
+    batch_exports.BatchExportLogViewSet,
+    "batch_export_logs",
+    ["team_id", "batch_export_id", "run_id"],
+)
 
 projects_router.register(r"warehouse_tables", table.TableViewSet, "project_warehouse_tables", ["team_id"])
 projects_router.register(

posthog/api/test/batch_exports/operations.py

@@ -104,3 +104,11 @@ def patch_batch_export(client, team_id, batch_export_id, new_batch_export_data):
         new_batch_export_data,
         content_type="application/json",
     )
+
+
+def get_batch_export_log_entries(client: TestClient, team_id: int, batch_export_id: str):
+    return client.get(f"/api/projects/{team_id}/batch_exports/{batch_export_id}/logs")
+
+
+def get_batch_export_run_log_entries(client: TestClient, team_id: int, batch_export_id: str, run_id):
+    return client.get(f"/api/projects/{team_id}/batch_exports/{batch_export_id}/runs/{run_id}/logs")