diff --git a/ee/urls.py b/ee/urls.py index b2cfcb73f438..421237471864 100644 --- a/ee/urls.py +++ b/ee/urls.py @@ -66,12 +66,6 @@ def extend_api_router() -> None: except NotRegistered: pass - from posthog.admin.admins.distinct_id_usage_admin import distinct_id_usage_view - from posthog.admin.admins.notebook_markdown_migration_admin import ( - notebook_markdown_migration_run_view, - notebook_markdown_migration_stats_view, - notebook_markdown_migration_view, - ) from posthog.admin.admins.tophog_admin import tophog_dashboard_view, tophog_restrictions_view from posthog.admin.console.urls import urlpatterns as console_urlpatterns @@ -91,7 +85,7 @@ def extend_api_router() -> None: ), path( "admin/distinct-id-usage/", - admin.site.admin_view(distinct_id_usage_view), + MovedToConsole.page_redirect("Distinct ID usage", "/admin/console/tools/distinct-id-usage/"), name="distinct-id-usage", ), path( @@ -126,18 +120,23 @@ def extend_api_router() -> None: ), path( "admin/notebook-markdown-migration/", - admin.site.admin_view(notebook_markdown_migration_view), + MovedToConsole.page_redirect( + "Markdown notebook migration", "/admin/console/tools/notebook-markdown-migration/" + ), name="notebook-markdown-migration", ), + # The old page fetched these JSON endpoints itself, so a bookmark of one lands on the tool page. path( "admin/notebook-markdown-migration/stats/", - admin.site.admin_view(notebook_markdown_migration_stats_view), - name="notebook-markdown-migration-stats", + MovedToConsole.page_redirect( + "Markdown notebook migration", "/admin/console/tools/notebook-markdown-migration/" + ), ), path( "admin/notebook-markdown-migration/run/", - admin.site.admin_view(notebook_markdown_migration_run_view), - name="notebook-markdown-migration-run", + MovedToConsole.page_redirect( + "Markdown notebook migration", "/admin/console/tools/notebook-markdown-migration/" + ), ), path( "admin/logout/", diff --git a/frontend/public/admin-console/crests/data-tools.svg b/frontend/public/admin-console/crests/data-tools.svg new file mode 100644 index 000000000000..eb6e4cb5503e --- /dev/null +++ b/frontend/public/admin-console/crests/data-tools.svg @@ -0,0 +1 @@ + \ No newline at end of file diff --git a/frontend/public/admin-console/crests/ingestion.svg b/frontend/public/admin-console/crests/ingestion.svg new file mode 100644 index 000000000000..2b116d225e70 --- /dev/null +++ b/frontend/public/admin-console/crests/ingestion.svg @@ -0,0 +1 @@ + \ No newline at end of file diff --git a/posthog/admin/admins/distinct_id_usage_admin.py b/posthog/admin/admins/distinct_id_usage_admin.py deleted file mode 100644 index 7fe2325980e2..000000000000 --- a/posthog/admin/admins/distinct_id_usage_admin.py +++ /dev/null @@ -1,466 +0,0 @@ -import logging -from datetime import UTC, datetime, timedelta - -from django import forms -from django.conf import settings -from django.contrib import admin, messages -from django.core.exceptions import PermissionDenied -from django.http import HttpResponseRedirect -from django.shortcuts import render -from django.urls import reverse -from django.utils.http import url_has_allowed_host_and_scheme - -from posthog.hogql.escape_sql import escape_clickhouse_identifier - -from posthog.clickhouse.client import sync_execute -from posthog.models.distinct_id_usage.sql import TABLE_BASE_NAME -from posthog.models.event_ingestion_restriction_config import EventIngestionRestrictionConfig, RestrictionType -from posthog.models.team.team import Team - -logger = logging.getLogger(__name__) - -QUERY_SETTINGS = {"max_execution_time": 30} - - -def _escaped_table() -> str: - db = escape_clickhouse_identifier(settings.CLICKHOUSE_DATABASE) - tbl = escape_clickhouse_identifier(TABLE_BASE_NAME) - return f"{db}.{tbl}" - - -def _query_high_usage( - dt_from: datetime, dt_to: datetime, threshold: int, min_events_threshold: int, distinct_id_min_events: int -) -> list[tuple]: - table = _escaped_table() - query = """ - WITH team_totals AS ( - SELECT team_id, sum(event_count) as total_events - FROM {table} - WHERE minute >= %(dt_from)s AND minute < %(dt_to)s - GROUP BY team_id - HAVING total_events >= %(min_events_threshold)s - ), - distinct_id_totals AS ( - SELECT team_id, distinct_id, sum(event_count) as event_count - FROM {table} - WHERE minute >= %(dt_from)s AND minute < %(dt_to)s - GROUP BY team_id, distinct_id - ) - SELECT d.team_id, d.distinct_id, d.event_count, t.total_events, - round(d.event_count * 100.0 / t.total_events, 2) as percentage - FROM distinct_id_totals d - JOIN team_totals t ON d.team_id = t.team_id - WHERE d.event_count * 100.0 / t.total_events >= %(threshold)s - AND d.event_count >= %(distinct_id_min_events)s - ORDER BY d.event_count DESC - LIMIT 100 - """.format(table=table) - return sync_execute( - query, - { - "dt_from": dt_from, - "dt_to": dt_to, - "threshold": threshold, - "min_events_threshold": min_events_threshold, - "distinct_id_min_events": distinct_id_min_events, - }, - settings=QUERY_SETTINGS, - ) - - -def _query_high_cardinality(dt_from: datetime, dt_to: datetime, threshold: int) -> list[tuple]: - table = _escaped_table() - query = """ - SELECT team_id, uniq(distinct_id) as distinct_id_count - FROM {table} - WHERE minute >= %(dt_from)s AND minute < %(dt_to)s - GROUP BY team_id - HAVING distinct_id_count >= %(threshold)s - ORDER BY distinct_id_count DESC - LIMIT 100 - """.format(table=table) - return sync_execute( - query, - { - "dt_from": dt_from, - "dt_to": dt_to, - "threshold": threshold, - }, - settings=QUERY_SETTINGS, - ) - - -def _query_bursts_by_distinct_id(dt_from: datetime, dt_to: datetime, threshold: int) -> list[tuple]: - table = _escaped_table() - query = """ - SELECT team_id, distinct_id, minute, event_count - FROM {table} - WHERE minute >= %(dt_from)s AND minute < %(dt_to)s - AND event_count >= %(threshold)s - ORDER BY event_count DESC - LIMIT 100 - """.format(table=table) - return sync_execute( - query, - { - "dt_from": dt_from, - "dt_to": dt_to, - "threshold": threshold, - }, - settings=QUERY_SETTINGS, - ) - - -def _query_bursts_by_team(dt_from: datetime, dt_to: datetime, threshold: int) -> list[tuple]: - table = _escaped_table() - query = """ - SELECT team_id, minute, sum(event_count) as total_events - FROM {table} - WHERE minute >= %(dt_from)s AND minute < %(dt_to)s - GROUP BY team_id, minute - HAVING total_events >= %(threshold)s - ORDER BY total_events DESC - LIMIT 100 - """.format(table=table) - return sync_execute( - query, - { - "dt_from": dt_from, - "dt_to": dt_to, - "threshold": threshold, - }, - settings=QUERY_SETTINGS, - ) - - -DATETIME_FORMAT = "%Y-%m-%dT%H:%M" - - -def _default_from(): - return (datetime.now(tz=UTC) - timedelta(hours=6)).strftime(DATETIME_FORMAT) - - -def _default_to(): - return datetime.now(tz=UTC).strftime(DATETIME_FORMAT) - - -class DistinctIdUsageForm(forms.Form): - datetime_from = forms.DateTimeField( - label="From", - widget=forms.DateTimeInput(attrs={"type": "datetime-local"}, format=DATETIME_FORMAT), - input_formats=[DATETIME_FORMAT], - ) - datetime_to = forms.DateTimeField( - label="To", - widget=forms.DateTimeInput(attrs={"type": "datetime-local"}, format=DATETIME_FORMAT), - input_formats=[DATETIME_FORMAT], - ) - - def clean(self): - cleaned = super().clean() - if cleaned is None: - return cleaned - dt_from = cleaned.get("datetime_from") - dt_to = cleaned.get("datetime_to") - if dt_from and dt_to: - if dt_from >= dt_to: - raise forms.ValidationError("'From' must be before 'To'.") - if (dt_to - dt_from).days > 7: - raise forms.ValidationError("Range cannot exceed 7 days (table TTL).") - return cleaned - - high_usage_percentage_threshold = forms.IntegerField( - initial=30, - min_value=1, - max_value=100, - label="High usage % threshold", - help_text="Percentage of team events from a single distinct_id", - ) - high_usage_min_events_threshold = forms.IntegerField( - initial=10_000, - min_value=1, - label="Min team events", - help_text="Minimum total team events to qualify", - ) - high_usage_distinct_id_min_events = forms.IntegerField( - initial=100_000, - min_value=1, - label="Min distinct_id events", - help_text="Minimum events from a distinct_id to trigger", - ) - high_cardinality_threshold = forms.IntegerField( - initial=1_000_000, - min_value=1, - label="High cardinality threshold", - help_text="Unique distinct_ids per team", - ) - burst_distinct_id_threshold = forms.IntegerField( - initial=100_000, - min_value=1, - label="Burst by distinct ID threshold", - help_text="Events per minute from a single (team, distinct_id)", - ) - burst_team_threshold = forms.IntegerField( - initial=500_000, - min_value=1, - label="Burst by team threshold", - help_text="Total events per minute from a single team", - ) - - -def _handle_add_to_restriction(request) -> None: - restriction_id = request.POST.get("restriction_id") - distinct_id = request.POST.get("distinct_id") - - if not restriction_id or not distinct_id: - messages.error(request, "Missing restriction_id or distinct_id.") - return - - try: - restriction = EventIngestionRestrictionConfig.objects.get(pk=restriction_id) - except EventIngestionRestrictionConfig.DoesNotExist: - messages.error(request, f"Restriction {restriction_id} not found.") - return - - label = f"{restriction.get_restriction_type_display()} restriction for {restriction.token[:16]}…" - if restriction.add_distinct_id(distinct_id): - messages.success(request, f"Added distinct_id to {label}") - else: - messages.info(request, f"distinct_id already in {label}") - - -def _handle_create_or_update_restriction(request) -> None: - token = request.POST.get("token") - restriction_type = request.POST.get("restriction_type") - distinct_id = request.POST.get("distinct_id") - - if not token or not restriction_type or not distinct_id: - messages.error(request, "Missing token, restriction_type, or distinct_id.") - return - - if restriction_type not in RestrictionType.values: - messages.error(request, f"Invalid restriction type: {restriction_type}") - return - - restriction, created, added = EventIngestionRestrictionConfig.add_distinct_id_for_token( - token, restriction_type, distinct_id - ) - label = restriction.get_restriction_type_display() - if created: - messages.success(request, f"Created {label} restriction for {token[:16]}…") - elif added: - messages.success(request, f"Added distinct_id to existing {label} restriction for {token[:16]}…") - else: - messages.info(request, f"distinct_id already in {label} restriction.") - - -def _handle_post_action(request) -> HttpResponseRedirect: - action = request.POST.get("action") - return_querystring = request.POST.get("return_querystring", "") - redirect_url = reverse("distinct-id-usage") - if return_querystring: - candidate = f"{redirect_url}?{return_querystring}" - if url_has_allowed_host_and_scheme(candidate, allowed_hosts={request.get_host()}): - redirect_url = candidate - - if action == "add_to_restriction": - _handle_add_to_restriction(request) - elif action == "create_or_update_restriction": - _handle_create_or_update_restriction(request) - else: - messages.error(request, f"Unknown action: {action}") - - return HttpResponseRedirect(redirect_url) - - -def distinct_id_usage_view(request): - if not request.user.is_staff: - raise PermissionDenied - - if request.method == "POST": - return _handle_post_action(request) - - initial = {"datetime_from": _default_from(), "datetime_to": _default_to()} - form = DistinctIdUsageForm(request.GET or None, initial=initial) - - high_usage_rows: list[dict] = [] - high_cardinality_rows: list[dict] = [] - burst_distinct_id_rows: list[dict] = [] - burst_team_rows: list[dict] = [] - errors: list[str] = [] - queried = False - - if form.is_valid(): - queried = True - dt_from = form.cleaned_data["datetime_from"] - dt_to = form.cleaned_data["datetime_to"] - - high_usage_raw: list[tuple] = [] - high_cardinality_raw: list[tuple] = [] - burst_distinct_id_raw: list[tuple] = [] - burst_team_raw: list[tuple] = [] - - try: - high_usage_raw = _query_high_usage( - dt_from, - dt_to, - threshold=form.cleaned_data["high_usage_percentage_threshold"], - min_events_threshold=form.cleaned_data["high_usage_min_events_threshold"], - distinct_id_min_events=form.cleaned_data["high_usage_distinct_id_min_events"], - ) - except Exception as e: - logger.exception("Distinct ID usage: high usage query failed") - errors.append(f"High usage query failed: {e}") - - try: - high_cardinality_raw = _query_high_cardinality( - dt_from, dt_to, threshold=form.cleaned_data["high_cardinality_threshold"] - ) - except Exception as e: - logger.exception("Distinct ID usage: high cardinality query failed") - errors.append(f"High cardinality query failed: {e}") - - try: - burst_distinct_id_raw = _query_bursts_by_distinct_id( - dt_from, dt_to, threshold=form.cleaned_data["burst_distinct_id_threshold"] - ) - except Exception as e: - logger.exception("Distinct ID usage: burst by distinct ID query failed") - errors.append(f"Burst by distinct ID query failed: {e}") - - try: - burst_team_raw = _query_bursts_by_team(dt_from, dt_to, threshold=form.cleaned_data["burst_team_threshold"]) - except Exception as e: - logger.exception("Distinct ID usage: burst by team query failed") - errors.append(f"Burst by team query failed: {e}") - - # Batch-lookup team API tokens - all_team_ids: set[int] = set() - for row in high_usage_raw: - all_team_ids.add(row[0]) - for row in high_cardinality_raw: - all_team_ids.add(row[0]) - for row in burst_distinct_id_raw: - all_team_ids.add(row[0]) - for row in burst_team_raw: - all_team_ids.add(row[0]) - - team_tokens: dict[int, str] = {} - if all_team_ids: - team_tokens = dict(Team.objects.filter(id__in=all_team_ids).values_list("id", "api_token")) - - # Fetch analytics pipeline restrictions indexed by token - # Each token can have multiple restriction types (skip_person, drop_event, etc.) - analytics_restrictions: dict[str, list[EventIngestionRestrictionConfig]] = {} - if team_tokens: - for r in EventIngestionRestrictionConfig.objects.filter( - token__in=team_tokens.values(), pipelines__contains=["analytics"] - ): - analytics_restrictions.setdefault(r.token, []).append(r) - - def _build_restriction_info(token: str, distinct_id: str | None = None) -> dict[str, str | bool]: - restrictions = analytics_restrictions.get(token, []) - if not restrictions: - return {"restriction_status": "none"} - - if distinct_id: - for r in restrictions: - if not r.distinct_ids or distinct_id in r.distinct_ids: - edit_url = reverse("admin:posthog_eventingestionrestrictionconfig_change", args=[r.pk]) - if not r.distinct_ids: - return { - "restriction_status": "covered_all", - "restriction_edit_url": edit_url, - "restriction_type": r.get_restriction_type_display(), - } - return { - "restriction_status": "covered", - "restriction_edit_url": edit_url, - "restriction_type": r.get_restriction_type_display(), - } - first = restrictions[0] - edit_url = reverse("admin:posthog_eventingestionrestrictionconfig_change", args=[first.pk]) - return { - "restriction_status": "not_covered", - "restriction_edit_url": edit_url, - "restriction_id": str(first.pk), - "restriction_type": first.get_restriction_type_display(), - } - - # No distinct_id context (high cardinality) - first = restrictions[0] - edit_url = reverse("admin:posthog_eventingestionrestrictionconfig_change", args=[first.pk]) - return { - "restriction_status": "exists", - "restriction_edit_url": edit_url, - "restriction_type": first.get_restriction_type_display(), - "restriction_count": str(len(restrictions)), - } - - for row in high_usage_raw: - team_id, distinct_id, event_count, total_team_events, percentage = row - token = team_tokens.get(team_id, "") - high_usage_rows.append( - { - "team_id": team_id, - "distinct_id": distinct_id, - "event_count": f"{event_count:,}", - "total_team_events": f"{total_team_events:,}", - "percentage": percentage, - "token": token, - **_build_restriction_info(token, distinct_id), - } - ) - - for row in high_cardinality_raw: - team_id, distinct_id_count = row - token = team_tokens.get(team_id, "") - high_cardinality_rows.append( - { - "team_id": team_id, - "distinct_id_count": f"{distinct_id_count:,}", - "token": token, - **_build_restriction_info(token), - } - ) - - for row in burst_distinct_id_raw: - team_id, distinct_id, minute, event_count = row - token = team_tokens.get(team_id, "") - burst_distinct_id_rows.append( - { - "team_id": team_id, - "distinct_id": distinct_id, - "minute": str(minute), - "event_count": f"{event_count:,}", - "token": token, - **_build_restriction_info(token, distinct_id), - } - ) - - for row in burst_team_raw: - team_id, minute, total_events = row - token = team_tokens.get(team_id, "") - burst_team_rows.append( - { - "team_id": team_id, - "minute": str(minute), - "event_count": f"{total_events:,}", - "token": token, - **_build_restriction_info(token), - } - ) - - context = { - **admin.site.each_context(request), - "title": "Distinct ID usage", - "form": form, - "queried": queried, - "high_usage_rows": high_usage_rows, - "high_cardinality_rows": high_cardinality_rows, - "burst_distinct_id_rows": burst_distinct_id_rows, - "burst_team_rows": burst_team_rows, - "errors": errors, - "return_querystring": request.GET.urlencode(), - } - return render(request, "admin/distinct_id_usage.html", context) diff --git a/posthog/admin/admins/notebook_markdown_migration_admin.py b/posthog/admin/admins/notebook_markdown_migration_admin.py deleted file mode 100644 index a184870683a4..000000000000 --- a/posthog/admin/admins/notebook_markdown_migration_admin.py +++ /dev/null @@ -1,129 +0,0 @@ -import json -from dataclasses import asdict -from typing import TYPE_CHECKING, Any, cast - -from django import forms -from django.contrib import admin -from django.http import HttpRequest, HttpResponse, JsonResponse -from django.shortcuts import render -from django.views.decorators.http import require_GET, require_POST - -from products.notebooks.backend.facade import api as notebooks_api - -if TYPE_CHECKING: - from posthog.models import User - -DEFAULT_NOTEBOOK_MIGRATION_BATCH_SIZE = 100 - - -class NotebookMarkdownMigrationForm(forms.Form): - team_id = forms.IntegerField( - required=False, - min_value=1, - help_text="Only count and convert notebooks for this team id. Leave blank to target all teams.", - ) - dry_run = forms.BooleanField( - required=False, - initial=True, - help_text="Preview the notebooks that would be converted without saving anything.", - ) - batch_size = forms.IntegerField( - required=False, - initial=DEFAULT_NOTEBOOK_MIGRATION_BATCH_SIZE, - min_value=1, - max_value=notebooks_api.MAX_NOTEBOOK_MIGRATION_BATCH_SIZE, - help_text=( - "Only process this many pending notebooks in one request. " - f"Use repeated batches for large scopes. Maximum {notebooks_api.MAX_NOTEBOOK_MIGRATION_BATCH_SIZE}." - ), - ) - - -def notebook_markdown_migration_view(request: HttpRequest) -> HttpResponse: - context = { - **admin.site.each_context(request), - "form": NotebookMarkdownMigrationForm(), - "title": "Markdown notebook migration", - } - return render(request, "admin/notebook_markdown_migration.html", context) - - -@require_GET -def notebook_markdown_migration_stats_view(request: HttpRequest) -> JsonResponse: - try: - team_id = _parse_team_id(request.GET.get("team_id")) - stats = notebooks_api.get_markdown_notebook_migration_stats(team_id) - except ValueError as err: - return JsonResponse({"error": str(err)}, status=400) - - return JsonResponse(asdict(stats)) - - -@require_POST -def notebook_markdown_migration_run_view(request: HttpRequest) -> JsonResponse: - try: - payload = json.loads(request.body.decode("utf-8") or "{}") - except json.JSONDecodeError: - return JsonResponse({"error": "Invalid JSON body"}, status=400) - if not isinstance(payload, dict): - return JsonResponse({"error": "JSON body must be an object"}, status=400) - - try: - team_id = _parse_team_id(payload.get("team_id")) - dry_run = _parse_bool(payload.get("dry_run", True)) - batch_size = _parse_batch_size(payload.get("batch_size")) - result = notebooks_api.migrate_notebooks_to_markdown( - user=cast("User", request.user), - team_id=team_id, - dry_run=dry_run, - batch_size=batch_size, - ) - except ValueError as err: - return JsonResponse({"error": str(err)}, status=400) - - return JsonResponse(_result_to_response(result)) - - -def _parse_team_id(raw_team_id: Any) -> int | None: - if raw_team_id is None or raw_team_id == "": - return None - try: - return int(raw_team_id) - except (TypeError, ValueError): - raise ValueError("Team id must be an integer") - - -def _parse_batch_size(raw_batch_size: Any) -> int | None: - if raw_batch_size is None or raw_batch_size == "": - return DEFAULT_NOTEBOOK_MIGRATION_BATCH_SIZE - try: - batch_size = int(raw_batch_size) - except (TypeError, ValueError): - raise ValueError("Batch size must be an integer") - if batch_size < 1: - raise ValueError("Batch size must be at least 1") - if batch_size > notebooks_api.MAX_NOTEBOOK_MIGRATION_BATCH_SIZE: - raise ValueError(f"Batch size must be {notebooks_api.MAX_NOTEBOOK_MIGRATION_BATCH_SIZE} or less") - return batch_size - - -def _parse_bool(raw_value: Any) -> bool: - if isinstance(raw_value, bool): - return raw_value - if isinstance(raw_value, str): - normalized_value = raw_value.strip().lower() - if normalized_value in ("true", "1", "yes", "on"): - return True - if normalized_value in ("false", "0", "no", "off"): - return False - raise ValueError("Dry run must be a boolean") - - -def _result_to_response(result: Any) -> dict[str, Any]: - data = asdict(result) - data["message"] = ( - f"Dry run previewed {result.converted} notebook(s) from this batch." - if result.dry_run - else f"Converted {result.converted} notebook(s) to markdown. {result.pending_after} pending." - ) - return data diff --git a/posthog/admin/console/distinct_id_usage_tool.py b/posthog/admin/console/distinct_id_usage_tool.py new file mode 100644 index 000000000000..32ca928154bd --- /dev/null +++ b/posthog/admin/console/distinct_id_usage_tool.py @@ -0,0 +1,375 @@ +from datetime import UTC, datetime, timedelta +from typing import Any + +from django import forms +from django.contrib import messages +from django.http import HttpRequest, HttpResponse, HttpResponseRedirect +from django.urls import reverse +from django.utils.http import urlencode + +from posthog.admin.console.panels import Owner, Tone +from posthog.admin.console.registry import console +from posthog.admin.console.tools import Cell, ConsoleTool, Notice, Page, Row, Table, ToolOperation +from posthog.models.distinct_id_usage.report import ( + DistinctIdUsage, + RestrictionRefused, + TokenRestriction, + UsageThresholds, +) +from posthog.models.event_ingestion_restriction_config import RestrictionType +from posthog.models.team.team import Team + +INGESTION = Owner(name="Ingestion", crest="ingestion") +TOOL_URL = "/admin/console/tools/distinct-id-usage/" +RESTRICT_URL = f"{TOOL_URL}restrict/" +DATETIME_FORMAT = "%Y-%m-%dT%H:%M" +DEFAULT_HOURS = 6 +# The table keeps 7 days of data, so a longer range reads nothing more. +MAX_RANGE_DAYS = 7 +RESTRICTION_CHOICES = [ + (RestrictionType.SKIP_PERSON_PROCESSING, "Skip person processing"), + (RestrictionType.DROP_EVENT_FROM_INGESTION, "Drop events"), + (RestrictionType.FORCE_OVERFLOW_FROM_INGESTION, "Force overflow"), +] + + +def _datetime_field(label: str) -> forms.DateTimeField: + return forms.DateTimeField( + label=label, + widget=forms.DateTimeInput(attrs={"type": "datetime-local"}, format=DATETIME_FORMAT), + input_formats=[DATETIME_FORMAT], + ) + + +class DistinctIdUsageForm(forms.Form): + datetime_from = _datetime_field("From (UTC)") + datetime_to = _datetime_field("To (UTC)") + high_usage_percentage_threshold = forms.IntegerField( + initial=30, + min_value=1, + max_value=100, + label="High usage share (%)", + help_text="Share of a team's events that come from one distinct ID.", + ) + high_usage_min_events_threshold = forms.IntegerField( + initial=10_000, + min_value=1, + label="Minimum team events", + help_text="Teams with fewer events in the range are left out of high usage.", + ) + high_usage_distinct_id_min_events = forms.IntegerField( + initial=100_000, + min_value=1, + label="Minimum distinct ID events", + help_text="Distinct IDs with fewer events in the range are left out of high usage.", + ) + high_cardinality_threshold = forms.IntegerField( + initial=1_000_000, + min_value=1, + label="High cardinality", + help_text="Unique distinct IDs per team.", + ) + burst_distinct_id_threshold = forms.IntegerField( + initial=100_000, + min_value=1, + label="Burst by distinct ID", + help_text="Events per minute from one distinct ID of a team.", + ) + burst_team_threshold = forms.IntegerField( + initial=500_000, + min_value=1, + label="Burst by team", + help_text="Events per minute from one team.", + ) + + def clean(self) -> dict[str, Any]: + cleaned = super().clean() or {} + dt_from = cleaned.get("datetime_from") + dt_to = cleaned.get("datetime_to") + if dt_from and dt_to: + if dt_from >= dt_to: + raise forms.ValidationError("From must be before To.") + if (dt_to - dt_from).days > MAX_RANGE_DAYS: + raise forms.ValidationError(f"Pick a range of {MAX_RANGE_DAYS} days or less. Older data is deleted.") + return cleaned + + def thresholds(self) -> UsageThresholds: + return UsageThresholds( + high_usage_percentage=self.cleaned_data["high_usage_percentage_threshold"], + high_usage_min_team_events=self.cleaned_data["high_usage_min_events_threshold"], + high_usage_min_distinct_id_events=self.cleaned_data["high_usage_distinct_id_min_events"], + high_cardinality=self.cleaned_data["high_cardinality_threshold"], + distinct_id_burst=self.cleaned_data["burst_distinct_id_threshold"], + team_burst=self.cleaned_data["burst_team_threshold"], + ) + + +REPORT_FIELDS = tuple(DistinctIdUsageForm.base_fields) + + +class RestrictDistinctIdForm(forms.Form): + token = forms.CharField(label="Project token", max_length=100) + # Capture keeps surrounding spaces and restrictions match exactly, so the ID is saved as typed. + distinct_id = forms.CharField(label="Distinct ID", max_length=450, strip=False) + restriction_type = forms.ChoiceField( + label="Restriction", choices=RESTRICTION_CHOICES, initial=RestrictionType.SKIP_PERSON_PROCESSING + ) + reason = forms.CharField(widget=forms.Textarea(attrs={"rows": 3}), help_text="Goes into the audit log.") + + def clean_token(self) -> str: + token = self.cleaned_data["token"].strip() + if not Team.objects.filter(api_token=token).exists(): + raise forms.ValidationError("No project has this token.") + return token + + +def run_report(request: HttpRequest, form: forms.Form | None) -> HttpResponse: + assert form is not None + # The report shows on the tool page, so its URL can be bookmarked and shared like the old page's. + query = urlencode({name: form.data[name] for name in REPORT_FIELDS}) + return HttpResponseRedirect(f"{TOOL_URL}?{query}") + + +def restrict_distinct_id(request: HttpRequest, form: forms.Form | None) -> HttpResponse | None: + assert form is not None + token = form.cleaned_data["token"] + try: + change = DistinctIdUsage.restrict( + token, form.cleaned_data["restriction_type"], form.cleaned_data["distinct_id"] + ) + except RestrictionRefused as refused: + messages.error(request, str(refused)) + return HttpResponse(status=400) + short_token = f"{token[:16]}…" + if change.created: + messages.success(request, f"Created a {change.label} restriction for {short_token} with this distinct ID.") + elif change.added: + messages.success(request, f"Added the distinct ID to the {change.label} restriction for {short_token}.") + else: + messages.info(request, f"The distinct ID is already in the {change.label} restriction for {short_token}.") + return None + + +def _restrict_url(request: HttpRequest, token: str, distinct_id: str, restriction_type: str | None = None) -> str: + # The report parameters stay in the link, so staff land back on the same report after the restriction. + query = request.GET.copy() + query["token"] = token + query["distinct_id"] = distinct_id + query.pop("restriction_type", None) + if restriction_type is not None: + query["restriction_type"] = restriction_type + return f"{RESTRICT_URL}?{query.urlencode()}" + + +def _change_url(restriction: TokenRestriction) -> str: + return reverse("admin:posthog_eventingestionrestrictionconfig_change", args=[restriction.id]) + + +def _team_cell(team_id: int) -> Cell: + return Cell(text=str(team_id), href=reverse("admin_console_team", kwargs={"pk": team_id})) + + +def _count(value: int) -> Cell: + return Cell(text=f"{value:,}") + + +def _distinct_id_restriction_cell( + request: HttpRequest, token: str, distinct_id: str, restrictions: tuple[TokenRestriction, ...] +) -> Cell: + if not restrictions: + return Cell(text="-", muted=True) + for restriction in restrictions: + if not restriction.distinct_ids: + return Cell(text=f"Applied to all: {restriction.label}", href=_change_url(restriction), tone=Tone.SUCCESS) + if distinct_id in restriction.distinct_ids: + return Cell(text=f"Applied: {restriction.label}", href=_change_url(restriction), tone=Tone.SUCCESS) + # These cells carry no tone, because the table renders a toned cell as a badge without its link. + restrictable_types = {choice for choice, _label in RESTRICTION_CHOICES} + for restriction in restrictions: + if restriction.restriction_type in restrictable_types: + return Cell( + text=f"Not in list: {restriction.label}", + href=_restrict_url(request, token, distinct_id, restriction.restriction_type), + ) + # The restrict form offers no redirect types, so the link leaves the type to staff. + return Cell(text=f"Not in list: {restrictions[0].label}", href=_restrict_url(request, token, distinct_id)) + + +def _team_restriction_cell(restrictions: tuple[TokenRestriction, ...]) -> Cell: + if not restrictions: + return Cell(text="-", muted=True) + first = restrictions[0] + more = len(restrictions) - 1 + return Cell(text=first.label + (f" and {more} more" if more else ""), href=_change_url(first)) + + +def _action_cell(request: HttpRequest, token: str, distinct_id: str) -> Cell: + if not token: + return Cell(text="No token", muted=True) + return Cell(text="Restrict", href=_restrict_url(request, token, distinct_id)) + + +def _report_initial(request: HttpRequest) -> dict[str, object]: + now = datetime.now(tz=UTC) + initial: dict[str, object] = { + "datetime_from": (now - timedelta(hours=DEFAULT_HOURS)).strftime(DATETIME_FORMAT), + "datetime_to": now.strftime(DATETIME_FORMAT), + } + initial.update({name: request.GET[name] for name in REPORT_FIELDS if name in request.GET}) + return initial + + +@console.register_tool +class DistinctIdUsageTool(ConsoleTool): + key = "distinct-id-usage" + title = "Distinct ID usage" + owner = INGESTION + group = "Data" + description = "Finds distinct IDs and teams that send unusually many events, and restricts their ingestion." + + def page(self, request: HttpRequest) -> Page: + if "datetime_from" not in request.GET: + return Page( + blocks=( + Notice( + text=( + "Run the report to find distinct IDs and teams that send unusually many events. " + f"The data covers the last {MAX_RANGE_DAYS} days." + ) + ), + ) + ) + form = DistinctIdUsageForm(request.GET) + if not form.is_valid(): + problems = " ".join(str(error) for errors in form.errors.values() for error in errors) + return Page(blocks=(Notice(text=f"The report did not run. {problems} Run the report again."),)) + + thresholds = form.thresholds() + report = DistinctIdUsage.report( + form.cleaned_data["datetime_from"], form.cleaned_data["datetime_to"], thresholds + ) + return Page( + blocks=( + *(Notice(text=error) for error in report.errors), + Notice( + text=( + f"High usage distinct IDs: {len(report.high_usage)}. One distinct ID sends at least " + f"{thresholds.high_usage_percentage}% of its team's events." + ) + ), + Table( + columns=("Team", "Distinct ID", "Restriction", "Events", "Team events", "Share", "Action"), + rows=tuple( + Row( + cells=( + _team_cell(row.team_id), + Cell(text=row.distinct_id, mono=True), + _distinct_id_restriction_cell(request, row.token, row.distinct_id, row.restrictions), + _count(row.event_count), + _count(row.team_event_count), + Cell(text=f"{row.percentage}%"), + _action_cell(request, row.token, row.distinct_id), + ) + ) + for row in report.high_usage + ), + empty_text="No distinct ID passes these thresholds.", + ), + Notice( + text=( + f"High cardinality teams: {len(report.high_cardinality)}. A team has at least " + f"{thresholds.high_cardinality:,} unique distinct IDs." + ) + ), + Table( + columns=("Team", "Restriction", "Unique distinct IDs"), + rows=tuple( + Row( + cells=( + _team_cell(row.team_id), + _team_restriction_cell(row.restrictions), + _count(row.distinct_id_count), + ) + ) + for row in report.high_cardinality + ), + empty_text="No team passes this threshold.", + ), + Notice( + text=( + f"Bursts by distinct ID: {len(report.distinct_id_bursts)}. One distinct ID sends at least " + f"{thresholds.distinct_id_burst:,} events in a minute." + ) + ), + Table( + columns=("Team", "Distinct ID", "Restriction", "Minute", "Events per minute", "Action"), + rows=tuple( + Row( + cells=( + _team_cell(row.team_id), + Cell(text=row.distinct_id, mono=True), + _distinct_id_restriction_cell(request, row.token, row.distinct_id, row.restrictions), + Cell(text=row.minute.strftime("%Y-%m-%d %H:%M")), + _count(row.event_count), + _action_cell(request, row.token, row.distinct_id), + ) + ) + for row in report.distinct_id_bursts + ), + empty_text="No distinct ID passes this threshold.", + ), + Notice( + text=( + f"Bursts by team: {len(report.team_bursts)}. One team sends at least " + f"{thresholds.team_burst:,} events in a minute." + ) + ), + Table( + columns=("Team", "Restriction", "Minute", "Events per minute"), + rows=tuple( + Row( + cells=( + _team_cell(row.team_id), + _team_restriction_cell(row.restrictions), + Cell(text=row.minute.strftime("%Y-%m-%d %H:%M")), + _count(row.event_count), + ) + ) + for row in report.team_bursts + ), + empty_text="No team passes this threshold.", + ), + ) + ) + + def operations(self, request: HttpRequest) -> list[ToolOperation]: + return [ + ToolOperation( + key="run", + label="Run report", + handler=run_report, + description="Reads the event counts of the time range, and lists what passes the thresholds.", + form_class=DistinctIdUsageForm, + form_initial=_report_initial(request), + ), + ToolOperation( + key="restrict", + label="Restrict a distinct ID", + handler=restrict_distinct_id, + danger=True, + description=( + "Adds the distinct ID to the project's ingestion restriction of this type. Creates the " + "restriction when the project has none." + ), + form_class=RestrictDistinctIdForm, + form_initial={ + name: request.GET[name] + for name in ("token", "distinct_id", "restriction_type") + if name in request.GET + }, + undo_hint=( + "To undo, remove the distinct ID from the restriction in Django admin, under Event ingestion " + "restriction configs." + ), + ), + ] diff --git a/posthog/admin/console/registry.py b/posthog/admin/console/registry.py index 22b6ea51e662..d99929e76411 100644 --- a/posthog/admin/console/registry.py +++ b/posthog/admin/console/registry.py @@ -86,6 +86,7 @@ def discover(self) -> None: import posthog.admin.console.usage_reports_tool # noqa: F401, PLC0415 — registers a core tool import posthog.admin.console.organization_panels # noqa: F401, PLC0415 — registers the core Organization panels import posthog.admin.console.events_retention_tool # noqa: F401, PLC0415 — registers a core tool + import posthog.admin.console.distinct_id_usage_tool # noqa: F401, PLC0415 — registers a core tool import posthog.admin.console.deletion_request_panels # noqa: F401, PLC0415 — registers the core deletion request panels autodiscover_modules("console") diff --git a/posthog/admin/test_admin_console.py b/posthog/admin/test_admin_console.py index 4eb6f21dce6a..6340fb446767 100644 --- a/posthog/admin/test_admin_console.py +++ b/posthog/admin/test_admin_console.py @@ -24,6 +24,7 @@ from posthog.admin.console.tools import Cell, ConsoleTool, Live, Notice, Page, Paging, Row, Table from posthog.csp_middleware import CSPMiddleware from posthog.models.data_deletion_request import DataDeletionRequest, RequestType +from posthog.models.event_ingestion_restriction_config import EventIngestionRestrictionConfig, RestrictionType from posthog.models.events_retention_config import OrganizationEventsRetentionConfig from posthog.models.organization import Organization from posthog.models.team.team import Team @@ -540,6 +541,97 @@ def test_events_retention_tool_lists_the_teams_of_an_organization_with_a_default assert "Organization default" in content assert "September 2025" in content + @patch("posthog.models.distinct_id_usage.report.DistinctIdUsageQueries") + def test_distinct_id_usage_report_keeps_the_old_query_and_marks_restrictions(self, queries: MagicMock) -> None: + EventIngestionRestrictionConfig.objects.create( + token=self.team.api_token, + restriction_type=RestrictionType.SKIP_PERSON_PROCESSING, + distinct_ids=["heavy-user"], + ) + EventIngestionRestrictionConfig.objects.create( + token=self.team.api_token, + restriction_type=RestrictionType.REDIRECT_TO_DLQ, + distinct_ids=["other-user"], + ) + redirected_team = Team.objects.create(organization=self.organization, name="Redirected") + EventIngestionRestrictionConfig.objects.create( + token=redirected_team.api_token, + restriction_type=RestrictionType.REDIRECT_TO_DLQ, + distinct_ids=["other-user"], + ) + queries.high_usage.return_value = [(self.team.pk, "heavy-user", 200_000, 300_000, 66.67)] + queries.high_cardinality.side_effect = RuntimeError("Timeout exceeded") + queries.distinct_id_bursts.return_value = [ + (self.team.pk, "bursty-user", datetime(2026, 11, 6, 10, 5, tzinfo=UTC), 150_000), + (redirected_team.pk, "bursty-user", datetime(2026, 11, 6, 10, 5, tzinfo=UTC), 150_000), + ] + queries.team_bursts.return_value = [] + query = ( + "datetime_from=2026-11-06T04%3A00&datetime_to=2026-11-06T10%3A00&high_usage_percentage_threshold=30" + "&high_usage_min_events_threshold=10000&high_usage_distinct_id_min_events=100000" + "&high_cardinality_threshold=1000000&burst_distinct_id_threshold=100000&burst_team_threshold=500000" + ) + + response = self.client.get(f"/admin/distinct-id-usage/?{query}", follow=True) + + assert response.redirect_chain == [(f"/admin/console/tools/distinct-id-usage/?{query}", 302)] + content = response.content.decode() + assert "Applied: Skip person processing" in content + assert "Not in list: Skip person processing" in content + assert "Not in list: Redirect to dlq" in content + # The restrict form offers no redirect types, so a redirect must never be prefilled. + assert "restriction_type=skip_person_processing" in content + assert "restriction_type=redirect_to_dlq" not in content + assert "The high cardinality query failed: Timeout exceeded" in content + assert "/admin/console/tools/distinct-id-usage/restrict/?" in content + + @parameterized.expand( + [ + ("creates_a_restriction", None, "Spam traffic", 302, ["heavy-user"]), + ("adds_to_the_existing_restriction", ["other-user"], "Spam traffic", 302, ["other-user", "heavy-user"]), + ("refuses_to_narrow_a_restriction_on_every_distinct_id", [], "Spam traffic", 400, []), + ("needs_a_reason", None, " ", 400, None), + ( + "refuses_a_restriction_of_other_pipelines", + ["other-user"], + "Spam traffic", + 400, + ["other-user"], + ("session_recordings",), + ), + ] + ) + def test_restrict_a_distinct_id( + self, + _name: str, + existing_distinct_ids: list[str] | None, + reason: str, + expected_status: int, + expected_distinct_ids: list[str] | None, + pipelines: tuple[str, ...] = ("analytics",), + ) -> None: + if existing_distinct_ids is not None: + EventIngestionRestrictionConfig.objects.create( + token=self.team.api_token, + restriction_type=RestrictionType.DROP_EVENT_FROM_INGESTION, + distinct_ids=existing_distinct_ids, + pipelines=list(pipelines), + ) + + response = self.client.post( + "/admin/console/tools/distinct-id-usage/restrict/", + { + "token": self.team.api_token, + "distinct_id": "heavy-user", + "restriction_type": RestrictionType.DROP_EVENT_FROM_INGESTION, + "reason": reason, + }, + ) + + assert response.status_code == expected_status + restriction = EventIngestionRestrictionConfig.objects.filter(token=self.team.api_token).first() + assert (restriction.distinct_ids if restriction else None) == expected_distinct_ids + @parameterized.expand( [ ("cached_config", ({"token": "phc_cached_config"}, "redis"), "phc_cached_config"), diff --git a/posthog/models/distinct_id_usage/report.py b/posthog/models/distinct_id_usage/report.py new file mode 100644 index 000000000000..0cf95a1305b9 --- /dev/null +++ b/posthog/models/distinct_id_usage/report.py @@ -0,0 +1,308 @@ +from collections.abc import Callable, Iterable +from datetime import datetime +from typing import Any + +from django.conf import settings + +import structlog + +from posthog.hogql.escape_sql import escape_clickhouse_identifier + +from posthog.clickhouse.client import sync_execute +from posthog.clickhouse.workload import Workload +from posthog.dataclasses import frozen +from posthog.models.distinct_id_usage.sql import TABLE_BASE_NAME +from posthog.models.event_ingestion_restriction_config import EventIngestionRestrictionConfig +from posthog.models.team.team import Team + +logger = structlog.get_logger(__name__) + +QUERY_SETTINGS = {"max_execution_time": 30} + + +@frozen +class UsageThresholds: + high_usage_percentage: int + high_usage_min_team_events: int + high_usage_min_distinct_id_events: int + high_cardinality: int + distinct_id_burst: int + team_burst: int + + +@frozen +class TokenRestriction: + """An analytics ingestion restriction on a project token. An empty `distinct_ids` restricts every distinct ID.""" + + id: str + restriction_type: str + label: str + distinct_ids: tuple[str, ...] + + +@frozen +class HighUsageDistinctId: + team_id: int + token: str + distinct_id: str + event_count: int + team_event_count: int + percentage: float + restrictions: tuple[TokenRestriction, ...] + + +@frozen +class HighCardinalityTeam: + team_id: int + token: str + distinct_id_count: int + restrictions: tuple[TokenRestriction, ...] + + +@frozen +class DistinctIdBurst: + team_id: int + token: str + distinct_id: str + minute: datetime + event_count: int + restrictions: tuple[TokenRestriction, ...] + + +@frozen +class TeamBurst: + team_id: int + token: str + minute: datetime + event_count: int + restrictions: tuple[TokenRestriction, ...] + + +@frozen +class DistinctIdUsageReport: + high_usage: tuple[HighUsageDistinctId, ...] + high_cardinality: tuple[HighCardinalityTeam, ...] + distinct_id_bursts: tuple[DistinctIdBurst, ...] + team_bursts: tuple[TeamBurst, ...] + # One line per query that failed. The other queries still report their rows. + errors: tuple[str, ...] + + +@frozen +class RestrictionChange: + label: str + created: bool + added: bool + + +class RestrictionRefused(Exception): + pass + + +class DistinctIdUsageQueries: + """Reads the distinct_id_usage table, which keeps per-minute event counts per distinct ID for 7 days.""" + + @staticmethod + def _execute(query: str, params: dict[str, Any]) -> list[tuple]: + # Staff wait on the console page for these results. + return sync_execute(query, params, settings=QUERY_SETTINGS, workload=Workload.ONLINE) + + @staticmethod + def _table() -> str: + db = escape_clickhouse_identifier(settings.CLICKHOUSE_DATABASE) + tbl = escape_clickhouse_identifier(TABLE_BASE_NAME) + return f"{db}.{tbl}" + + @classmethod + def high_usage(cls, dt_from: datetime, dt_to: datetime, thresholds: UsageThresholds) -> list[tuple]: + query = """ + WITH team_totals AS ( + SELECT team_id, sum(event_count) as total_events + FROM {table} + WHERE minute >= %(dt_from)s AND minute < %(dt_to)s + GROUP BY team_id + HAVING total_events >= %(min_events_threshold)s + ), + distinct_id_totals AS ( + SELECT team_id, distinct_id, sum(event_count) as event_count + FROM {table} + WHERE minute >= %(dt_from)s AND minute < %(dt_to)s + GROUP BY team_id, distinct_id + ) + SELECT d.team_id, d.distinct_id, d.event_count, t.total_events, + round(d.event_count * 100.0 / t.total_events, 2) as percentage + FROM distinct_id_totals d + JOIN team_totals t ON d.team_id = t.team_id + WHERE d.event_count * 100.0 / t.total_events >= %(threshold)s + AND d.event_count >= %(distinct_id_min_events)s + ORDER BY d.event_count DESC + LIMIT 100 + """.format(table=cls._table()) + return cls._execute( + query, + { + "dt_from": dt_from, + "dt_to": dt_to, + "threshold": thresholds.high_usage_percentage, + "min_events_threshold": thresholds.high_usage_min_team_events, + "distinct_id_min_events": thresholds.high_usage_min_distinct_id_events, + }, + ) + + @classmethod + def high_cardinality(cls, dt_from: datetime, dt_to: datetime, thresholds: UsageThresholds) -> list[tuple]: + query = """ + SELECT team_id, uniq(distinct_id) as distinct_id_count + FROM {table} + WHERE minute >= %(dt_from)s AND minute < %(dt_to)s + GROUP BY team_id + HAVING distinct_id_count >= %(threshold)s + ORDER BY distinct_id_count DESC + LIMIT 100 + """.format(table=cls._table()) + return cls._execute(query, {"dt_from": dt_from, "dt_to": dt_to, "threshold": thresholds.high_cardinality}) + + @classmethod + def distinct_id_bursts(cls, dt_from: datetime, dt_to: datetime, thresholds: UsageThresholds) -> list[tuple]: + query = """ + SELECT team_id, distinct_id, minute, event_count + FROM {table} + WHERE minute >= %(dt_from)s AND minute < %(dt_to)s + AND event_count >= %(threshold)s + ORDER BY event_count DESC + LIMIT 100 + """.format(table=cls._table()) + return cls._execute(query, {"dt_from": dt_from, "dt_to": dt_to, "threshold": thresholds.distinct_id_burst}) + + @classmethod + def team_bursts(cls, dt_from: datetime, dt_to: datetime, thresholds: UsageThresholds) -> list[tuple]: + query = """ + SELECT team_id, minute, sum(event_count) as total_events + FROM {table} + WHERE minute >= %(dt_from)s AND minute < %(dt_to)s + GROUP BY team_id, minute + HAVING total_events >= %(threshold)s + ORDER BY total_events DESC + LIMIT 100 + """.format(table=cls._table()) + return cls._execute(query, {"dt_from": dt_from, "dt_to": dt_to, "threshold": thresholds.team_burst}) + + +class DistinctIdUsage: + """Finds distinct IDs and teams that send unusually many events, and restricts their ingestion.""" + + @staticmethod + def _analytics_restrictions(tokens: Iterable[str]) -> dict[str, tuple[TokenRestriction, ...]]: + by_token: dict[str, list[TokenRestriction]] = {} + for restriction in EventIngestionRestrictionConfig.objects.filter( + token__in=list(tokens), pipelines__contains=["analytics"] + ): + by_token.setdefault(restriction.token, []).append( + TokenRestriction( + id=str(restriction.pk), + restriction_type=restriction.restriction_type, + label=restriction.get_restriction_type_display().capitalize(), + distinct_ids=tuple(restriction.distinct_ids or ()), + ) + ) + return {token: tuple(restrictions) for token, restrictions in by_token.items()} + + @classmethod + def report(cls, dt_from: datetime, dt_to: datetime, thresholds: UsageThresholds) -> DistinctIdUsageReport: + errors: list[str] = [] + + def run(name: str, query: Callable[[datetime, datetime, UsageThresholds], list[tuple]]) -> list[tuple]: + try: + return query(dt_from, dt_to, thresholds) + except Exception as error: + logger.exception("distinct_id_usage_query_failed", query=name) + errors.append(f"The {name} query failed: {error}") + return [] + + high_usage = run("high usage", DistinctIdUsageQueries.high_usage) + high_cardinality = run("high cardinality", DistinctIdUsageQueries.high_cardinality) + distinct_id_bursts = run("burst by distinct ID", DistinctIdUsageQueries.distinct_id_bursts) + team_bursts = run("burst by team", DistinctIdUsageQueries.team_bursts) + + team_ids = {row[0] for rows in (high_usage, high_cardinality, distinct_id_bursts, team_bursts) for row in rows} + tokens: dict[int, str] = ( + dict(Team.objects.filter(id__in=team_ids).values_list("id", "api_token")) if team_ids else {} + ) + restrictions = cls._analytics_restrictions(tokens.values()) if tokens else {} + + def token_of(team_id: int) -> str: + return tokens.get(team_id, "") + + def restrictions_of(team_id: int) -> tuple[TokenRestriction, ...]: + return restrictions.get(token_of(team_id), ()) + + return DistinctIdUsageReport( + high_usage=tuple( + HighUsageDistinctId( + team_id=team_id, + token=token_of(team_id), + distinct_id=distinct_id, + event_count=event_count, + team_event_count=team_event_count, + percentage=percentage, + restrictions=restrictions_of(team_id), + ) + for team_id, distinct_id, event_count, team_event_count, percentage in high_usage + ), + high_cardinality=tuple( + HighCardinalityTeam( + team_id=team_id, + token=token_of(team_id), + distinct_id_count=distinct_id_count, + restrictions=restrictions_of(team_id), + ) + for team_id, distinct_id_count in high_cardinality + ), + distinct_id_bursts=tuple( + DistinctIdBurst( + team_id=team_id, + token=token_of(team_id), + distinct_id=distinct_id, + minute=minute, + event_count=event_count, + restrictions=restrictions_of(team_id), + ) + for team_id, distinct_id, minute, event_count in distinct_id_bursts + ), + team_bursts=tuple( + TeamBurst( + team_id=team_id, + token=token_of(team_id), + minute=minute, + event_count=event_count, + restrictions=restrictions_of(team_id), + ) + for team_id, minute, event_count in team_bursts + ), + errors=tuple(errors), + ) + + @staticmethod + def restrict(token: str, restriction_type: str, distinct_id: str) -> RestrictionChange: + existing = EventIngestionRestrictionConfig.objects.filter( + token=token, restriction_type=restriction_type + ).first() + if existing is not None and "analytics" not in (existing.pipelines or []): + # One row per token and type, so a row scoped to other pipelines cannot also cover analytics. + raise RestrictionRefused( + f"The {existing.get_restriction_type_display().capitalize()} restriction of this project applies to " + f"{', '.join(existing.pipelines or [])} only, not analytics, so nothing changed. " + "Add analytics to its pipelines first." + ) + if existing is not None and not existing.distinct_ids: + # An empty list matches every distinct ID, so adding one would narrow the restriction to that one. + raise RestrictionRefused( + f"The {existing.get_restriction_type_display().capitalize()} restriction of this project applies to " + "every distinct ID already, so nothing changed." + ) + restriction, created, added = EventIngestionRestrictionConfig.add_distinct_id_for_token( + token, restriction_type, distinct_id + ) + return RestrictionChange( + label=restriction.get_restriction_type_display().capitalize(), created=created, added=added + ) diff --git a/posthog/templates/admin/distinct_id_usage.html b/posthog/templates/admin/distinct_id_usage.html deleted file mode 100644 index 6d3efb5b5506..000000000000 --- a/posthog/templates/admin/distinct_id_usage.html +++ /dev/null @@ -1,393 +0,0 @@ -{% extends "admin/base_site.html" %} -{% load i18n admin_urls static %} -{% block extrastyle %} - {{ block.super }} - - - - -{% endblock %} -{% if not is_popup %} - {% block breadcrumbs %} - - {% endblock %} -{% endif %} -{% block content %} -
- {% if errors %} - {% for error in errors %} -

{{ error }}

- {% endfor %} - {% endif %} -
-
-

Configuration

-
- - {{ form.datetime_from }} - - {{ form.datetime_to }} -
- {{ form.non_field_errors }} -

Thresholds

-
- {% for field in form %} - {% if field.name != "datetime_from" and field.name != "datetime_to" %} -
- - {{ field }} - {% if field.help_text %}

{{ field.help_text }}

{% endif %} - {{ field.errors }} -
- {% endif %} - {% endfor %} -
-
-
- -
-
- {% if queried %} - -
- - - {% if high_usage_rows %} - - - - - - - - - - - - - - {% for row in high_usage_rows %} - - - - - - - - - - - {% endfor %} - - {% else %} - - - - - - {% endif %} -
High usage distinct IDs {{ high_usage_rows|length }}
Team IDTokenDistinct IDRestrictionsEvent countTeam total%Actions
{{ row.team_id }}{{ row.token|truncatechars:16 }}Copied{{ row.distinct_id|truncatechars:40 }}Copied - {% if row.restriction_status == "covered" %} - ✓ Applied ({{ row.restriction_type }}) Edit - {% elif row.restriction_status == "covered_all" %} - ✓ Applied (all) ({{ row.restriction_type }}) Edit - {% elif row.restriction_status == "not_covered" %} - ✗ Not in list ({{ row.restriction_type }}) -
- {% csrf_token %} - - - - - -
- {% else %} - — - {% endif %} -
{{ row.event_count }}{{ row.total_team_events }}{{ row.percentage }}% - {% if row.token %} -
- {% csrf_token %} - - - - - - -
-
- {% csrf_token %} - - - - - - -
-
- {% csrf_token %} - - - - - - -
- {% else %} - No token - {% endif %} -
No high usage distinct IDs found with current thresholds.
-
- -
- - - {% if high_cardinality_rows %} - - - - - - - - - - {% for row in high_cardinality_rows %} - - - - - - - {% endfor %} - - {% else %} - - - - - - {% endif %} -
High cardinality teams {{ high_cardinality_rows|length }}
Team IDTokenRestrictionsUnique distinct IDs
{{ row.team_id }}{{ row.token|truncatechars:16 }}Copied - {% if row.restriction_status == "exists" %} - {{ row.restriction_type }}{% if row.restriction_count > 1 %} (+{{ row.restriction_count|add:"-1" }}){% endif %} Edit - {% else %} - — - {% endif %} - {{ row.distinct_id_count }}
No high cardinality teams found with current thresholds.
-
- -
- - - {% if burst_distinct_id_rows %} - - - - - - - - - - - - - {% for row in burst_distinct_id_rows %} - - - - - - - - - - {% endfor %} - - {% else %} - - - - - - {% endif %} -
Burst events by distinct ID {{ burst_distinct_id_rows|length }}
Team IDTokenDistinct IDRestrictionsMinuteEvents/minActions
{{ row.team_id }}{{ row.token|truncatechars:16 }}Copied{{ row.distinct_id|truncatechars:40 }}Copied - {% if row.restriction_status == "covered" %} - ✓ Applied ({{ row.restriction_type }}) Edit - {% elif row.restriction_status == "covered_all" %} - ✓ Applied (all) ({{ row.restriction_type }}) Edit - {% elif row.restriction_status == "not_covered" %} - ✗ Not in list ({{ row.restriction_type }}) -
- {% csrf_token %} - - - - - -
- {% else %} - — - {% endif %} -
{{ row.minute }}{{ row.event_count }} - {% if row.token %} -
- {% csrf_token %} - - - - - - -
-
- {% csrf_token %} - - - - - - -
-
- {% csrf_token %} - - - - - - -
- {% else %} - No token - {% endif %} -
No burst events by distinct ID found with current thresholds.
-
- -
- - - {% if burst_team_rows %} - - - - - - - - - - - {% for row in burst_team_rows %} - - - - - - - - {% endfor %} - - {% else %} - - - - - - {% endif %} -
Burst events by team {{ burst_team_rows|length }}
Team IDTokenRestrictionsMinuteEvents/min
{{ row.team_id }}{{ row.token|truncatechars:16 }}Copied - {% if row.restriction_status == "exists" %} - {{ row.restriction_type }}{% if row.restriction_count > 1 %} (+{{ row.restriction_count|add:"-1" }}){% endif %} Edit - {% else %} - — - {% endif %} - {{ row.minute }}{{ row.event_count }}
No burst events by team found with current thresholds.
-
- {% endif %} -
-{% endblock %} diff --git a/posthog/templates/admin/notebook_markdown_migration.html b/posthog/templates/admin/notebook_markdown_migration.html deleted file mode 100644 index 218793330e39..000000000000 --- a/posthog/templates/admin/notebook_markdown_migration.html +++ /dev/null @@ -1,286 +0,0 @@ -{% extends "admin/base_site.html" %} -{% load i18n admin_urls static %} -{% block extrastyle %} - {{ block.super }} - -{% endblock %} -{% block content %} -
-
-

Migration status

-

Counts include every notebook row in scope, including internal, account-linked, and deleted notebooks.

-
- - Loading… -
-
-
- Total notebooks - - -
-
- Converted - - -
-
- Pending - - -
-
-
-
-

Run migration

-

- This rewrites rich notebooks into the markdown notebook format using the backend converter, - increments the notebook version, publishes an update event, and writes a notebook history entry - with the before/after content diff. -

-
- {% csrf_token %} -
- - {{ form.team_id }} -

{{ form.team_id.help_text }}

-
-
- -

{{ form.dry_run.help_text }}

-
-
- - {{ form.batch_size }} -

{{ form.batch_size.help_text }}

-
-
- -
-
-
-
-
-
- -{% endblock %} diff --git a/products/notebooks/backend/console.py b/products/notebooks/backend/console.py new file mode 100644 index 000000000000..dd267bb19812 --- /dev/null +++ b/products/notebooks/backend/console.py @@ -0,0 +1,179 @@ +from typing import cast + +from django import forms +from django.contrib import messages +from django.http import HttpRequest, HttpResponse, HttpResponseRedirect + +from posthog.admin.console.panels import Owner +from posthog.admin.console.registry import console +from posthog.admin.console.tools import Cell, ConsoleTool, Notice, Page, Row, Table, ToolOperation +from posthog.models.user import User +from posthog.utils import pluralize + +from products.notebooks.backend.facade import api +from products.notebooks.backend.facade.contracts import MarkdownNotebookMigrationResult + +DATA_TOOLS = Owner(name="Data tools", crest="data-tools") +TOOL_URL = "/admin/console/tools/notebook-markdown-migration/" +DEFAULT_BATCH_SIZE = 100 +PREVIEW_CHARACTERS = 200 + + +class ConvertNotebooksForm(forms.Form): + team_id = forms.IntegerField( + label="Team ID", required=False, min_value=1, help_text="Leave empty to convert notebooks of every team." + ) + batch_size = forms.IntegerField( + initial=DEFAULT_BATCH_SIZE, + min_value=1, + max_value=api.MAX_NOTEBOOK_MIGRATION_BATCH_SIZE, + help_text="Pending notebooks converted in one run. Run it again for the next batch.", + ) + reason = forms.CharField(widget=forms.Textarea(attrs={"rows": 3}), help_text="Goes into the audit log.") + + +def _scope(team_id: int | None) -> str: + return f"team {team_id}" if team_id is not None else "all teams" + + +def _migrate(request: HttpRequest, form: forms.Form, dry_run: bool) -> MarkdownNotebookMigrationResult: + return api.migrate_notebooks_to_markdown( + user=cast(User, request.user), + team_id=form.cleaned_data["team_id"], + dry_run=dry_run, + batch_size=form.cleaned_data["batch_size"], + ) + + +def preview_conversion(request: HttpRequest, form: forms.Form) -> Page: + try: + result = _migrate(request, form, dry_run=True) + except ValueError as error: + form.add_error("team_id", str(error)) + return Page(blocks=()) + if result.converted == 0 and result.errored == 0: + form.add_error(None, f"No notebook of {_scope(result.team_id)} waits for conversion.") + return Page(blocks=()) + + summary = ( + f"Converts {pluralize(result.converted, 'notebook')} of the {result.pending_before} pending in " + f"{_scope(result.team_id)}. {result.already_converted} of {result.total} are markdown already." + ) + if result.converted > len(result.previews): + summary += f" The table shows the first {len(result.previews)}." + blocks: list[Notice | Table] = [ + Notice(text=summary), + Table( + columns=("Notebook", "Title", "Version", "Markdown"), + rows=tuple( + Row( + cells=( + Cell(text=preview.short_id, mono=True), + Cell(text=preview.title or "Untitled", muted=not preview.title), + Cell(text=str(preview.before_version)), + Cell(text=preview.markdown_preview[:PREVIEW_CHARACTERS], mono=True), + ) + ) + for preview in result.previews + ), + ), + ] + if result.errored: + blocks.append( + Notice( + text=f"{pluralize(result.errored, 'notebook')} cannot be converted and stay as they are: " + + "; ".join(result.errors) + ) + ) + return Page(blocks=tuple(blocks)) + + +def convert_notebooks(request: HttpRequest, form: forms.Form | None) -> HttpResponse | Page | None: + assert form is not None + try: + result = _migrate(request, form, dry_run=False) + except ValueError as error: + messages.error(request, str(error)) + return HttpResponse(status=400) + messages.success( + request, + f"Converted {pluralize(result.converted, 'notebook')} to markdown. " + f"{result.pending_after} still pending in {_scope(result.team_id)}.", + ) + if result.skipped: + messages.info(request, f"Skipped {pluralize(result.skipped, 'notebook')} that another change converted first.") + if result.errored: + messages.warning( + request, + f"{pluralize(result.errored, 'notebook')} could not be converted: " + "; ".join(result.errors), + ) + # Back to the scope that just ran, so the next batch's form keeps the team. + if result.team_id is None: + return HttpResponseRedirect(TOOL_URL) + return HttpResponseRedirect(f"{TOOL_URL}?team_id={result.team_id}") + + +def _team_id(request: HttpRequest) -> int | None: + raw = request.GET.get("team_id", "").strip() + return int(raw) if raw.isdigit() else None + + +@console.register_tool +class NotebookMarkdownMigrationTool(ConsoleTool): + key = "notebook-markdown-migration" + title = "Markdown notebook migration" + owner = DATA_TOOLS + group = "Data" + description = "Converts rich notebooks to markdown notebooks in batches, after a preview." + + def page(self, request: HttpRequest) -> Page: + team_id = _team_id(request) + try: + stats = api.get_markdown_notebook_migration_stats(team_id) + except ValueError as error: + return Page(blocks=(Notice(text=f"{error}.", link_text="All teams", href=TOOL_URL),)) + + notice = ( + "Converts rich notebooks to the markdown notebook format. Each converted notebook gets a new version, " + "open editors get the update, and the notebook history keeps the content from before. The counts " + "include internal, account-linked and deleted notebooks." + ) + return Page( + blocks=( + Notice(text=notice, link_text="All teams", href=TOOL_URL) + if team_id is not None + else Notice(text=notice), + Table( + columns=("Scope", "Notebooks", "Markdown", "Pending"), + rows=( + Row( + cells=( + Cell(text=_scope(team_id).capitalize()), + Cell(text=f"{stats.total:,}"), + Cell(text=f"{stats.converted:,}"), + Cell(text=f"{stats.pending:,}"), + ) + ), + ), + ), + ) + ) + + def operations(self, request: HttpRequest) -> list[ToolOperation]: + team_id = _team_id(request) + return [ + ToolOperation( + key="convert", + label="Convert notebooks", + handler=convert_notebooks, + preview=preview_conversion, + danger=True, + description="Rewrites one batch of pending notebooks in the markdown format.", + form_class=ConvertNotebooksForm, + form_initial={"team_id": team_id} if team_id is not None else None, + undo_hint=( + "PostHog Admin cannot convert a notebook back. Each notebook's history keeps its content from " + "before the conversion." + ), + ) + ] diff --git a/products/notebooks/backend/test/test_markdown_migration.py b/products/notebooks/backend/test/test_markdown_migration.py index 85862337fbd4..7b0e133d51b3 100644 --- a/products/notebooks/backend/test/test_markdown_migration.py +++ b/products/notebooks/backend/test/test_markdown_migration.py @@ -1,3 +1,4 @@ +import re from datetime import timedelta from posthog.test.base import BaseTest @@ -533,3 +534,49 @@ def test_apply_scopes_mention_label_lookup_to_notebook_organization(self) -> Non assert f'@member' in notebook.text_content assert "Leaked" not in notebook.text_content assert "leaked@example.com" not in notebook.text_content + + +class TestNotebookMarkdownMigrationTool(BaseTest): + TOOL_URL = "/admin/console/tools/notebook-markdown-migration/" + + def setUp(self) -> None: + super().setUp() + self.user.is_staff = True + self.user.save() + self.client.force_login(self.user) + + def test_old_stats_url_lands_on_the_tool_with_its_team(self) -> None: + response = self.client.get(f"/admin/notebook-markdown-migration/stats/?team_id={self.team.id}") + + assert response.status_code == 302 + assert response["Location"] == f"{self.TOOL_URL}?team_id={self.team.id}" + + @patch("products.notebooks.backend.markdown_collab.publish_notebook_update") + def test_convert_previews_without_changes_then_converts_on_confirm(self, _mock_publish) -> None: + notebook = Notebook.objects.create( + team=self.team, + title="Convert from the console", + content={"type": "doc", "content": [{"type": "paragraph", "content": [{"type": "text", "text": "body"}]}]}, + version=3, + ) + convert_url = f"{self.TOOL_URL}convert/" + + preview = self.client.post( + convert_url, {"team_id": str(self.team.id), "batch_size": "10", "reason": "Markdown rollout"} + ) + notebook.refresh_from_db() + + assert preview.status_code == 200 + assert "Convert from the console" in preview.content.decode() + assert notebook.version == 3 + + token = re.search(r'name="confirm_token" value="([^"]+)"', preview.content.decode()) + assert token is not None + with self.captureOnCommitCallbacks(execute=True): + response = self.client.post(convert_url, {"confirm_token": token.group(1)}) + notebook.refresh_from_db() + + assert response.status_code == 302 + assert response["Location"] == f"{self.TOOL_URL}?team_id={self.team.id}" + assert notebook.version == 4 + assert notebook.content["content"][0]["type"] == "ph-markdown-notebook"