Repository: PostHog/posthog
Pull request: Closes #1059: Async processing of Event -> Action mapping
Closes #1059 
- Add Env Var `ASYNC_EVENT_ACTION_MAPPING` for deactivating inline Event Action lookups to prevent gnarly query generation resulting in Postgres OOM
- Setup a Celery Beat task to do Action -> Event mapping every minute on all Actions (we may want to tune this or set this as the env var parameter)

Already test coverage on `Action.calculate_events()` in a few places
https://github.com/PostHog/posthog/blob/da5a48c33d059be9d5c0cb4259524560454e2df8/posthog/api/test/test_action.py#L84


Problem statement:
Adding actions causes Postgres out of memory with partitions
**Describe the bug**

Report from OpenRent that adding a simple $autocapture action causes Postgres to run out of memory. I think this has something to do with the fact that they've got things partitioned.

**To Reproduce**
1. TBD
2. 
3. 

**Expected behavior**

It should work.

**Screenshots**



**Hosted or self hosted?:**
- self-hosted

**Additional context**

Hints from issue discussion:
@timgl Will I be looking into this tomorrow? Do we have a repro case? :)
The issue from what I've seen and been able to reproduce is:
- An instance that is [partitioned](https://posthog.com/docs/deployment/scaling-posthog)
- Has high-ish volume, somewhere around 3 RPS
- Has an action defined. This seems to happen with both pageview actions and $autocapture ("frontend element") ones, but I can consistently reproduce with a single autocapture event, even if that event doesn't match anything.
The errors that appear all indicate out of memory on the postgres database:
![image](https://user-images.githubusercontent.com/1727427/85517932-09ddaf80-b600-11ea-8be4-14c9942709ea.png)
Removing the one action solves these errors
- We do [some caching](https://github.com/PostHog/posthog/blob/master/posthog/models/event.py#L294) of the generation of the queries, so the issue here is the actual query that function generates causes the database to run out of memory quickly.

I think the solution is finding a way of making that query more optimized.

Thanks for the overview @timgl. I dug into this briefly as well:

So the issue seems to come from **data ingestion** running into errors on a high level.

On a high level, what happens during ingestion seems to be the following:
1. API receives a request, it gets queued in celery for the worker node to pick up (posthog/api/event.py)
2. Worker node processes event data, creates/updates appropriate users and events (posthog/tasks/process_event.py)
3. `Event.create` gets called, which first creates the event itself (posthog/models/event.py EventManager.create)
4. After saving, it finds actions associated with these events. (posthog/models/event.py Event#events)
5. From this it calculates relations to be added, adds them to the database (posthog/models/event.py EventManager.create)
6. Also notifies slack if there's a webhook

I think there's two ways to approach this:

### Specific problem: `Event.create` OOM

From above, we think the OOM error is caused by step 4, which is storing stuff in memory as well as querying the database. Note that `TEAM_ACTION_QUERY_CACHE` for example is never cleared, which might end up taking up a significant chunk of memory.

### High-level problem: does this flow scale?

There's two potential bottlenecks during ingestion:

1. API throughput
2. The database

Step 4 is fundamentally gonna require looking through the database in some fashion, in ways that get slower as users get more data. I think 2 is gonna become an issue much faster than 1 due to this.

Given that steps 4-6 (gonna call them post-processing henceforth) require tying up a lot of db resources, I'd argue it's worth doing them asynchronously from the rest of the ingestion. That would open us up to some further optimizations like:

1. Dedicated workers which can calculate event -> action relations more efficiently (by using e.g. in-memory caching). By using in-memory caching, this could ease the load on the db as well.
1a. Dedicated workers which are partitioned by e.g. team for hosted solution or by id range otherwise.
2. Scaling ingestion and "post-postprocessing" separately.
3. Post-processing in batches

The downside here is that the solution is more complex (scalability inevitably is) and that not all ingested data is always ready to query.

### My thoughts

I think it's briefly worth taking a look at the specific problem (e.g. via heapdumps, etc), I think most of the benefit will come from tackling the high-level problem here.

### Questions:

1) Is Celery set up to retry tasks when they fail or would they currently go to the dustbin?
2) Since action.events relation many-to-many is hard to scale, what are the arguments for it to be made "virtual"? I.e. instead of figuring it out at write-time, figure it out when querying. Not familiar enough with the schema yet to answer this.
Re the 'specific problem': the memory issue isn't on the worker side but definitely on the database side, as the errors are Postgres errors. So it's the query that we generate that's causing the issue.

Those queries are already called in the workers and they happen constantly, so I don't think handling them separately is going to work. Instead I'd be good to understand (using explain etc) _why_ these queries are using so much mem, because each query only calls a single event.

To answer your specific questions
1. I think we don't have any retrying implemented, could be worth looking at but would only make this specific problem worse
2. Querying at run time is what we had originally and become really slow, which is why we moved to m2m. It'd be awesome to find a way to do this quickly during run time but not sure that's realistic?
> Re the 'specific problem': the memory issue isn't on the worker side but definitely on the database side, as the errors are Postgres errors. So it's the query that we generate that's causing the issue.

Thanks for the additional context. :+1: 

> Those queries are already called in the workers and they happen constantly, 

I realize that and what I described above tried to take that into consideration.

So my goal with the "high level problem" was to bring out that there's a fundamental scalability issue trying to do ingestion not at 3 RPS but at 300 or 3000 RPS.

The latter is very well possible - but I posit the "post-processing" steps here would become the bottleneck due to how the underlying dbs work.

Given that this issue is at least tangentially related to this issue I thought it best to bring it up as well and to propose a solution.

In this case I think a solution to the high-level problem could help with the specific issue - being able to cache in memory, doing less expensive queries. That said, I think _both_ sides of this issue are worth taking a look at.
Discussed this with Tim. As said before - we'll tackle the "specific problem" for now. Happy I have some thoughts regarding scaling written down though :) Only later realized that this might not apply as badly as I thought since not every unique event type creates an action :sweat_smile: 
@timgl How is OpenRent hosting posthog? Do we know what sort of memory/cpu limits their setup has, are worker/api/postgres co-hosted, how many actions they have?
I managed to create a repro case.

The issue is not directly related with partitioning itself, but is a function of autocapture + no of actions defined on autocapture + amount of data. Partitioning can however make the issue worse: queries may hit more partitions than needed.

## How to reproduce the issue

1. Get a dump of prod data on ph team (tim can help here)
2. Open shell via `python manage.py shell_plus --print-sql` (depends on PR https://github.com/PostHog/posthog/pull/1090). You might need to also set SHELL_PLUS_PRINT_SQL_TRUNCATE if you want to see full queries (https://django-extensions.readthedocs.io/en/latest/shell_plus.html?highlight=print-sql#sql-queries)
3. Run the following command(s):

```python
from posthog.tasks.process_event import process_event; import datetime

process_event(**{'distinct_id': '2020-06-22::0', 'ip': '127.0.0.1', 'site_url': 'http://0.0.0.0:8000', 'data': {'timestamp': '2020-06-24T14:07:36.150Z', 'event': '$autocapture', 'properties': {'$lib': 'posthog-node', '$lib_version': '1.0.3', 'some_prop': 'foobar', 'idx': 55}, 'type': 'capture', 'library': 'posthog-node', 'library_version': '1.0.3', 'distinct_id': '2020-06-22::0'}, 'team_id': 1, 'now': datetime.datetime(2020, 6, 24, 14, 7, 36), 'sent_at': None})
```

You should see in the console sql queries the process event does, with one of the queries taking ~40 seconds. Dump of the SQL it generated for me: https://gist.github.com/macobo/036d44d9143acd15937d87d7d6dcd7ab

Note that this query has 334 FROM clauses in various subselects.

## Background, what's causing the issue

So the query originates from the following lines in models/event.py: https://github.com/PostHog/posthog/blob/f6df41a/posthog/models/event.py#L348-L355

So lines 332 to 342 are trying to load all actions user has defined with the same name as the sent event ($autocapture in our case). 

Next steps on line 348-355 are trying to create a column for every one of these actions has a yes/no response if this action truly matches the event or not. This is done via a subquery generated by `query_db_by_action` function.

This logic breaks down when:
1. There are a lot of actions which match the sent event
2. There is a considerable amount of data already in the db
3. Query generated does not work as intended

All 3 are the case here:
1. $autocapture has a lot of actions defined - by nature. 
    - Postgres in turn needs to do a lot of work running a bunch of subqueries and then joining the results of which back together. I'd say this is a relatively tricky task for pg.
2. prod database has considerable size, though not massive by any means currently.
3. $autocapture queries can get tricky 
    - see e.g. https://github.com/PostHog/posthog/issues/1089 for an example where the generated subquery is much more complicated than expected. 
    - See also the generated explain in gist above - some of these subqueries are far from simple. See e.g. https://gist.github.com/macobo/036d44d9143acd15937d87d7d6dcd7ab#file-explain-sql-L424-L641 which is one of "Deploy to X" actions which gets turned into quite a complicated subquery, most of it to do with css selector matching.

#### Q: What about caching, does it help here?

There's a caching layer implemented in posthog/models/event.py, but this caching does not affect this problem - it only deals with query building, not executing it. 

## Where to from here?

This seems like a NotTrivialProblem™ - a side-effect of the current schema choices.

Some thoughts/ideas/suggestions:

1. Pulling out separate workers for action <-> event matching still seems like a great idea for the reasons outlined above. Instead of doing the matching in complicated SQL queries which both the ORM and postgres struggle with, this would mean we can do the matching in memory and load actions from the db lazily - easing the load considerably.
2. Loading actions into memory within the current pipeline should work well and also lend itself well to memoization/caching.
3. Optimizing CSS selector matching will help both here and elsewhere. The current solution tricky currently both in terms of SQL it generates and performance. Few ideas: it might be possible to extract a PL/pgSQL function for doing this matching or turn the css selector into a regex.
4. Reconsider actions completely. Currently they are used for both "annotating" common filters, making some queries more efficient (thanks to the 1:n relationship with events) and slack webhooks. Perhaps instead of trying to solve all 3 issues at once, instead make actions focus on the filtering part and use other tricks (indexing/queues/workers) for the latter two.

Commit message: add ACTION_EVENT_MAPPING_INTERVAL_MINUTES as env var

Patch to review:
diff --git a/latest_migrations.manifest b/latest_migrations.manifest
index b7e6ef3b08904..4da1564b3869a 100644
--- a/latest_migrations.manifest
+++ b/latest_migrations.manifest
@@ -1,6 +1,6 @@
 admin: 0003_logentry_add_action_flag_choices
 auth: 0011_update_proxy_permissions
 contenttypes: 0002_remove_content_type_name
-posthog: 0068_auto_20200629_1322
+posthog: 0069_auto_20200713_2125
 sessions: 0001_initial
 social_django: 0008_partial_timestamp
diff --git a/posthog/api/action.py b/posthog/api/action.py
index b53315af7453d..21ee93ca16001 100644
--- a/posthog/api/action.py
+++ b/posthog/api/action.py
@@ -1,5 +1,4 @@
 from django.db.models.expressions import Subquery
-from posthog.models import person
 from posthog.models import (
     Event,
     Team,
@@ -46,14 +45,10 @@
 import pandas as pd
 import datetime
 import json
-import pytz
 import copy
 import numpy as np
 from dateutil.relativedelta import relativedelta
-import dateutil
 from .person import PersonSerializer
-from urllib.parse import urlsplit
-from django.core.cache import cache
 from posthog.decorators import cached_function, TRENDS_ENDPOINT
 
 FREQ_MAP = {"minute": "60S", "hour": "H", "day": "D", "week": "W", "month": "M"}
diff --git a/posthog/celery.py b/posthog/celery.py
index 5fc007452d5f9..0b3d89735c114 100644
--- a/posthog/celery.py
+++ b/posthog/celery.py
@@ -10,6 +10,7 @@
 from datetime import datetime
 from dateutil import parser
 
+
 # set the default Django settings module for the 'celery' program.
 os.environ.setdefault("DJANGO_SETTINGS_MODULE", "posthog.settings")
 
@@ -42,6 +43,15 @@ def setup_periodic_tasks(sender, **kwargs):
     sender.add_periodic_task(15 * 60, calculate_cohort.s(), name="debug")
     sender.add_periodic_task(600, check_cached_items.s(), name="check dashboard items")
 
+    action_event_mapping_interval_minutes = settings.ACTION_EVENT_MAPPING_INTERVAL_MINUTES
+    if settings.ASYNC_EVENT_ACTION_MAPPING:
+        sender.add_periodic_task(
+            (60 * action_event_mapping_interval_minutes),
+            calculate_event_action_mappings.s(),
+            name="calculate event action mappings",
+            expires=(60 * action_event_mapping_interval_minutes),
+        )
+
 
 @app.task
 def redis_heartbeat():
@@ -56,6 +66,13 @@ def update_event_partitions():
         )
 
 
+@app.task
+def calculate_event_action_mappings():
+    from posthog.tasks.calculate_action import calculate_actions_from_last_calculation
+
+    calculate_actions_from_last_calculation()
+
+
 @app.task
 def calculate_cohort():
     from posthog.tasks.calculate_cohort import calculate_cohorts
diff --git a/posthog/migrations/0069_auto_20200713_2125.py b/posthog/migrations/0069_auto_20200713_2125.py
new file mode 100644
index 0000000000000..384851e1fb8f7
--- /dev/null
+++ b/posthog/migrations/0069_auto_20200713_2125.py
@@ -0,0 +1,36 @@
+# Generated by Django 3.0.7 on 2020-07-13 21:25
+
+from django.db import migrations, models
+import django.utils.timezone
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("posthog", "0068_auto_20200629_1322"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="action",
+            name="last_calculated_at",
+            field=models.DateTimeField(blank=True, default=django.utils.timezone.now),
+        ),
+        migrations.AddField(
+            model_name="event",
+            name="created_at",
+            field=models.DateTimeField(auto_now_add=True, default=None),
+            preserve_default=False,
+        ),
+        migrations.AlterField(
+            model_name="user",
+            name="toolbar_mode",
+            field=models.CharField(
+                blank=True,
+                choices=[("default", "default"), ("toolbar", "toolbar")],
+                default="default",
+                max_length=200,
+                null=True,
+            ),
+        ),
+    ]
diff --git a/posthog/models/action.py b/posthog/models/action.py
index 1fbd216b9ac7f..7796dc34aa55c 100644
--- a/posthog/models/action.py
+++ b/posthog/models/action.py
@@ -1,5 +1,8 @@
+import datetime
+
 from django.db import models, connection, transaction
 from django.core.exceptions import EmptyResultSet
+from django.utils import timezone
 from .user import User
 from sentry_sdk import capture_exception
 
@@ -10,28 +13,60 @@ class Meta:
             models.Index(fields=["team_id", "-updated_at"]),
         ]
 
-    def calculate_events(self):
+    def calculate_events(self, start=None, end=None):
+        recalculate_all = False
+        if start is None and end is None:
+            recalculate_all = True
+        if start is None:
+            start = datetime.date(1990, 1, 1)
+        if end is None:
+            end = timezone.now() + datetime.timedelta(days=1)
+
+        calculated_at = timezone.now()
         self.is_calculating = True
         self.save()
         from .event import Event
 
         try:
-            event_query, params = Event.objects.query_db_by_action(self).only("pk").query.sql_with_params()
+            if recalculate_all:
+                event_query, params = Event.objects.query_db_by_action(self).only("pk").query.sql_with_params()
+            else:
+                event_query, params = (
+                    Event.objects.query_db_by_action(self, start=start, end=end).only("pk").query.sql_with_params()
+                )
+
         except EmptyResultSet:
+            self.last_calculated_at = calculated_at
             self.is_calculating = False
             self.save()
             self.events.all().delete()
             return
 
-        query = """
-        DELETE FROM "posthog_action_events" WHERE "action_id" = {};
-        INSERT INTO "posthog_action_events" ("action_id", "event_id")
-        {}
-        ON CONFLICT DO NOTHING
-        """.format(
-            self.pk, event_query.replace("SELECT ", "SELECT {}, ".format(self.pk), 1)
+        query = """DELETE FROM "posthog_action_events" WHERE "action_id" = {}""".format(self.pk)
+
+        period_delete_query = """AND "event_id" in
+                       (SELECT id
+                        FROM posthog_event
+                        WHERE "created_at" >= '{}'
+                        AND "created_at" < '{}');
+                """.format(
+            start.isoformat(), end.isoformat()
         )
 
+        insert_query = """INSERT INTO "posthog_action_events" ("action_id", "event_id")
+                        {}
+                    ON CONFLICT DO NOTHING
+                 """.format(
+            event_query.replace("SELECT ", "SELECT {}, ".format(self.pk), 1)
+        )
+
+        if not recalculate_all:
+            query += period_delete_query
+        else:
+            query += ";"
+
+        query += insert_query
+
         cursor = connection.cursor()
         with transaction.atomic():
             try:
@@ -40,6 +75,7 @@ def calculate_events(self):
                 capture_exception()
 
         self.is_calculating = False
+        self.last_calculated_at = calculated_at
         self.save()
 
     name: models.CharField = models.CharField(max_length=400, null=True, blank=True)
@@ -51,6 +87,7 @@ def calculate_events(self):
     post_to_slack: models.BooleanField = models.BooleanField(default=False)
     is_calculating: models.BooleanField = models.BooleanField(default=False)
     updated_at: models.DateTimeField = models.DateTimeField(auto_now=True)
+    last_calculated_at: models.DateTimeField = models.DateTimeField(default=timezone.now, blank=True)
 
     def __str__(self):
         return self.name
diff --git a/posthog/models/event.py b/posthog/models/event.py
index 617719cc8c457..ad2e1d72d9645 100644
--- a/posthog/models/event.py
+++ b/posthog/models/event.py
@@ -1,3 +1,4 @@
+from django.conf import settings
 from django.db import models, transaction
 from django.db.models import (
     Exists,
@@ -144,6 +145,15 @@ def filter_by_event(self, action_step):
             return {}
         return {"event": action_step.event}
 
+    def filter_by_period(self, start, end):
+        if not start and not end:
+            return {}
+        if not start:
+            return {"created_at__lte": end}
+        if not end:
+            return {"created_at__gte": start}
+        return {"created_at__gte": start, "created_at__lte": end}
+
     def add_person_id(self, team_id: int):
         return self.annotate(
             person_id=Subquery(
@@ -153,7 +163,7 @@ def add_person_id(self, team_id: int):
             )
         )
 
-    def query_db_by_action(self, action, order_by="-timestamp") -> models.QuerySet:
+    def query_db_by_action(self, action, order_by="-timestamp", start=None, end=None) -> models.QuerySet:
         events = self
         any_step = Q()
         steps = action.steps.all()
@@ -167,7 +177,8 @@ def query_db_by_action(self, action, order_by="-timestamp") -> models.QuerySet:
                     Filter(data={"properties": step.properties}).properties_to_Q(team_id=action.team_id),
                     pk=OuterRef("id"),
                     **self.filter_by_event(step),
-                    **self.filter_by_element(model_to_dict(step), team_id=action.team_id)
+                    **self.filter_by_element(model_to_dict(step), team_id=action.team_id),
+                    **self.filter_by_period(start, end)
                 )
                 .only("id")
             )
@@ -244,17 +255,21 @@ def create(self, site_url: Optional[str] = None, *args: Any, **kwargs: Any):
                     ).hash
             event = super().create(*args, **kwargs)
 
-            should_post_to_slack = False
-            relations = []
-            for action in event.actions:
-                relations.append(action.events.through(action_id=action.pk, event_id=event.pk))
-                if action.post_to_slack:
-                    should_post_to_slack = True
-
-            Action.events.through.objects.bulk_create(relations, ignore_conflicts=True)
-            team = kwargs.get("team", event.team)
-            if should_post_to_slack and team and team.slack_incoming_webhook:
-                post_event_to_slack.delay(event.pk, site_url)
+            # Matching actions to events can get very expensive to do as events are streaming in
+            # In a few cases we have had it OOM Postgres with the query it is running
+            # Short term solution is to have this be configurable to be run in batch
+            if not settings.ASYNC_EVENT_ACTION_MAPPING:
+                should_post_to_slack = False
+                relations = []
+                for action in event.actions:
+                    relations.append(action.events.through(action_id=action.pk, event_id=event.pk))
+                    if action.post_to_slack:
+                        should_post_to_slack = True
+
+                Action.events.through.objects.bulk_create(relations, ignore_conflicts=True)
+                team = kwargs.get("team", event.team)
+                if should_post_to_slack and team and team.slack_incoming_webhook:
+                    post_event_to_slack.delay(event.pk, site_url)
 
             return event
 
@@ -354,6 +369,7 @@ def actions(self) -> List:
         filtered_actions = [action for action in actions if getattr(event, "action_{}".format(action.pk))]
         return filtered_actions
 
+    created_at: models.DateTimeField = models.DateTimeField(auto_now_add=True, blank=True)
     objects: EventManager = EventManager.as_manager()  # type: ignore
     team: models.ForeignKey = models.ForeignKey(Team, on_delete=models.CASCADE)
     event: models.CharField = models.CharField(max_length=200, null=True, blank=True)
diff --git a/posthog/settings.py b/posthog/settings.py
index 47a5573d67f24..e7fc5919f0153 100644
--- a/posthog/settings.py
+++ b/posthog/settings.py
@@ -79,6 +79,13 @@ def get_bool_from_env(name: str, default_value: bool) -> bool:
     USE_X_FORWARDED_HOST = True
     SECURE_PROXY_SSL_HEADER = ("HTTP_X_FORWARDED_PROTO", "https")
 
+ASYNC_EVENT_ACTION_MAPPING = False
+
+if get_bool_from_env("ASYNC_EVENT_ACTION_MAPPING", False):
+    ASYNC_EVENT_ACTION_MAPPING = True
+
+ACTION_EVENT_MAPPING_INTERVAL_MINUTES = float(os.getenv("ACTION_EVENT_MAPPING_INTERVAL_MINUTES", 0.5))
+
 # IP block settings
 ALLOWED_IP_BLOCKS = get_list(os.environ.get("ALLOWED_IP_BLOCKS", ""))
 TRUSTED_PROXIES = os.environ.get("TRUSTED_PROXIES", False)
diff --git a/posthog/tasks/calculate_action.py b/posthog/tasks/calculate_action.py
index 2d237b638a734..caa7ab22f29a3 100644
--- a/posthog/tasks/calculate_action.py
+++ b/posthog/tasks/calculate_action.py
@@ -4,6 +4,7 @@
 import logging
 import time
 
+
 logger = logging.getLogger(__name__)
 
 
@@ -13,3 +14,11 @@ def calculate_action(action_id: int) -> None:
     action = Action.objects.get(pk=action_id)
     action.calculate_events()
     logger.info("Calculating action {} took {:.2f} seconds".format(action.pk, (time.time() - start_time)))
+
+
+def calculate_actions_from_last_calculation() -> None:
+    actions = Action.objects.filter(deleted=False).only("pk")
+    for action in actions:
+        start_time = time.time()
+        action.calculate_events(start=action.last_calculated_at)
+        logger.info("Calculating action {} took {:.2f} seconds".format(action.pk, (time.time() - start_time)))
diff --git a/posthog/test/test_migration_0039.py b/posthog/test/test_migration_0039.py
index 62da20ecfc79a..f154b04612a56 100644
--- a/posthog/test/test_migration_0039.py
+++ b/posthog/test/test_migration_0039.py
@@ -41,55 +41,3 @@ def setUp(self):
             executor.migrate(migrate_to)
 
         self.apps = executor.loader.project_state(migrate_to).apps  # type: ignore
-
-
-class TagsTestCase(TestMigrations):
-
-    migrate_from = "0038_migrate_actions_to_precalculate_events"
-    migrate_to = "0039_populate_event_ip_property"
-
-    def setUpBeforeMigration(self, apps):
-        max_event_count = 1000000
-        max_batch_size = 10000
-        default_event_count = 100
-        default_batch_size = 10
-        Event = apps.get_model("posthog", "Event")
-        Team = apps.get_model("posthog", "Team")
-        team = Team.objects.create()
-
-        Event.objects.bulk_create(
-            [Event(team=team, event="$autocapture", ip="127.0.0.1") for i in range(default_event_count)],
-            default_batch_size,
-        )
-
-        Event.objects.bulk_create(
-            [
-                Event(
-                    team=team,
-                    event="$prefilled_properties",
-                    distinct_id="2",
-                    ip="192.2.2.1",
-                    properties={"$os": "MacOS"},
-                )
-                for i in range(default_event_count)
-            ],
-            default_batch_size,
-        )
-
-        null_ip_event = Event.objects.create(team=team, event="$null_ip")
-
-    def test_ip_migrated(self):
-        Event = apps.get_model("posthog", "Event")
-
-        events = Event.objects.all()
-        for e in events:
-            if e.event == "$prefilled_properties":
-                self.assertEqual(e.properties.get("$ip"), "192.2.2.1")
-                self.assertEqual(e.properties.get("$ip"), "192.2.2.1")
-                self.assertEqual(e.properties.get("$os"), "MacOS")
-
-            if e.event == "$autocapture":
-                self.assertEqual(e.properties.get("$ip"), "127.0.0.1")
-                self.assertEqual(e.properties.get("$ip"), "127.0.0.1")
-            if e.event == "$null_ip":
-                self.assertEqual(e.properties.get("$ip"), None)


Code review comments:
diff_hunk: @@ -42,6 +43,15 @@ def setup_periodic_tasks(sender, **kwargs):
     sender.add_periodic_task(15 * 60, calculate_cohort.s(), name="debug")
     sender.add_periodic_task(600, check_cached_items.s(), name="check dashboard items")
 
+    action_event_mapping_interval_minutes = settings.ACTION_EVENT_MAPPING_INTERVAL_MINUTES
path: posthog/celery.py
text: @user1:
I don't think this should be an env variable. Let's just pick a sensible default. Every 10 minutes seems fine to me

diff_hunk: @@ -354,6 +369,7 @@ def actions(self) -> List:
         filtered_actions = [action for action in actions if getattr(event, "action_{}".format(action.pk))]
         return filtered_actions
 
+    created_at: models.DateTimeField = models.DateTimeField(auto_now_add=True, blank=True)
path: posthog/models/event.py
text: @user1:
this should have `, null=True`