Hotfix for missing session recording events

Temporarily solves #2927,
though the error will rear it's head again with plugin-based ingestion.

Opened https://github.com/PostHog/posthog/issues/3068 for long-term
solution.

ee/clickhouse/client.py

@@ -104,11 +104,11 @@ def cache_sync_execute(query, args=None, redis_client=None, ttl=CACHE_TTL):
             redis_client.set(key, _serialize(result), ex=ttl)
             return result
 
-    def sync_execute(query, args=None):
+    def sync_execute(query, args=None, **kwargs):
         start_time = time()
         try:
             with ch_sync_pool.get_client() as client:
-                result = client.execute(query, args)
+                result = client.execute(query, args, **kwargs)
         finally:
             execution_time = time() - start_time
             if settings.SHELL_PLUS_PRINT_SQL:

ee/clickhouse/models/session_recording_event.py

@@ -1,16 +1,22 @@
 import datetime
 import json
+import logging
 import uuid
-from typing import Dict, List, Optional, Tuple, Union
+from typing import Union
 
-from dateutil.parser import isoparse
-from django.utils import timezone
+from sentry_sdk import capture_exception
 
+from ee.clickhouse.client import sync_execute
 from ee.clickhouse.models.util import cast_timestamp_or_now
 from ee.clickhouse.sql.session_recording_events import INSERT_SESSION_RECORDING_EVENT_SQL
 from ee.kafka_client.client import ClickhouseProducer
 from ee.kafka_client.topics import KAFKA_SESSION_RECORDING_EVENTS
 
+logger = logging.getLogger(__name__)
+
+MAX_KAFKA_MESSAGE_LENGTH = 800_000
+MAX_INSERT_LENGTH = 15_000_000
+
 
 def create_session_recording_event(
     uuid: uuid.UUID,
@@ -22,15 +28,22 @@ def create_session_recording_event(
 ) -> str:
     timestamp = cast_timestamp_or_now(timestamp)
 
+    snapshot_data_json = json.dumps(snapshot_data)
     data = {
         "uuid": str(uuid),
         "team_id": team_id,
         "distinct_id": distinct_id,
         "session_id": session_id,
-        "snapshot_data": json.dumps(snapshot_data),
+        "snapshot_data": snapshot_data_json,
         "timestamp": timestamp,
         "created_at": timestamp,
     }
-    p = ClickhouseProducer()
-    p.produce(sql=INSERT_SESSION_RECORDING_EVENT_SQL, topic=KAFKA_SESSION_RECORDING_EVENTS, data=data)
+    if len(snapshot_data_json) <= MAX_KAFKA_MESSAGE_LENGTH:
+        p = ClickhouseProducer()
+        p.produce(sql=INSERT_SESSION_RECORDING_EVENT_SQL, topic=KAFKA_SESSION_RECORDING_EVENTS, data=data)
+    elif len(snapshot_data_json) <= MAX_INSERT_LENGTH:
+        sync_execute(INSERT_SESSION_RECORDING_EVENT_SQL, data, settings={"max_query_size": MAX_INSERT_LENGTH})
+    else:
+        capture_exception(Exception(f"Session recording event data too large - {len(snapshot_data_json)}"))
+
     return str(uuid)