feat(events_stream): Expose event extras as individual fields in events_stream views (DENG-9546) (#8361)

* feat(events_stream): Expose event extras as individual fields in `events_stream` views (DENG-9546).

* fix(events_stream): Use `UNION ALL BY NAME` because the columns may not always be in the same order.

* Correct typo in `sql_generators/glean_usage/events_stream.py`.

Co-authored-by: Ben Wu <[email protected]>

---------

Co-authored-by: Ben Wu <[email protected]>

Author: sean-rose

sql_generators/glean_usage/common.py

@@ -17,7 +17,8 @@
 from bigquery_etl.schema.stable_table_schema import get_stable_table_schemas
 from bigquery_etl.util.common import get_table_dir, render, write_sql
 
-APP_LISTINGS_URL = "https://probeinfo.telemetry.mozilla.org/v2/glean/app-listings"
+PROBEINFO_URL = "https://probeinfo.telemetry.mozilla.org"
+APP_LISTINGS_URL = f"{PROBEINFO_URL}/v2/glean/app-listings"
 PATH = Path(os.path.dirname(__file__))
 
 # added as the result of baseline checks being added to the template,
@@ -174,6 +175,28 @@ def get_app_info() -> dict[str, list[dict]]:
     return app_info
 
 
+@cache
+def get_glean_repositories() -> list[dict]:
+    """Return a list of the Glean repositories."""
+    resp = requests.get(f"{PROBEINFO_URL}/glean/repositories")
+    resp.raise_for_status()
+    return resp.json()
+
+
+def get_glean_app_pings(v1_name: str) -> dict[str, dict]:
+    """Return a dictionary of the Glean app's pings."""
+    resp = requests.get(f"{PROBEINFO_URL}/glean/{v1_name}/pings")
+    resp.raise_for_status()
+    return resp.json()
+
+
+def get_glean_app_metrics(v1_name: str) -> dict[str, dict]:
+    """Return a dictionary of the Glean app's metrics."""
+    resp = requests.get(f"{PROBEINFO_URL}/glean/{v1_name}/metrics")
+    resp.raise_for_status()
+    return resp.json()
+
+
 class GleanTable:
     """Represents a generated Glean table."""
 

sql_generators/glean_usage/event_monitoring_live.py

@@ -6,12 +6,12 @@
 from pathlib import Path
 from typing import List, Set
 
-import requests
-
 from bigquery_etl.config import ConfigLoader
 from bigquery_etl.schema.stable_table_schema import get_stable_table_schemas
 from sql_generators.glean_usage.common import (
     GleanTable,
+    get_glean_app_metrics,
+    get_glean_app_pings,
     get_table_dir,
     render,
     table_names_from_baseline,
@@ -22,9 +22,6 @@
 TARGET_DATASET_CROSS_APP = "monitoring"
 PREFIX = "event_monitoring"
 PATH = Path(os.path.dirname(__file__))
-GLEAN_APP_BASE_URL = "https://probeinfo.telemetry.mozilla.org/glean/{app_name}"
-METRICS_INFO_URL = f"{GLEAN_APP_BASE_URL}/metrics"
-PING_INFO_URL = f"{GLEAN_APP_BASE_URL}/pings"
 
 
 class EventMonitoringLive(GleanTable):
@@ -54,15 +51,11 @@ def _get_tables_with_events(
     ) -> Set[str]:
         """Get tables for the given app that receive event type metrics."""
         pings = set()
-        metrics_resp = requests.get(METRICS_INFO_URL.format(app_name=v1_name))
-        metrics_resp.raise_for_status()
-        metrics_json = metrics_resp.json()
+        metrics_json = get_glean_app_metrics(v1_name)
 
         min_pings = set()
         if skip_min_ping:
-            ping_resp = requests.get(PING_INFO_URL.format(app_name=v1_name))
-            ping_resp.raise_for_status()
-            ping_json = ping_resp.json()
+            ping_json = get_glean_app_pings(v1_name)
             min_pings = {
                 name
                 for name, info in ping_json.items()

sql_generators/glean_usage/events_stream.py

@@ -1,9 +1,18 @@
 """Generate events stream queries for Glean apps."""
 
 import re
+from collections import defaultdict
+from datetime import date, datetime, timedelta
+from functools import cache
+from typing import Optional
 
 from bigquery_etl.config import ConfigLoader
-from sql_generators.glean_usage.common import GleanTable, ping_has_metrics
+from sql_generators.glean_usage.common import (
+    GleanTable,
+    get_glean_app_metrics,
+    get_glean_repositories,
+    ping_has_metrics,
+)
 
 TARGET_TABLE_ID = "events_stream_v1"
 PREFIX = "events_stream"
@@ -85,6 +94,9 @@ def generate_per_app_id(
                 app_id_info["bq_dataset_family"], unversioned_table_name
             ),
             "slice_by_sample_id": slice_by_sample_id,
+            "extras_by_type": get_glean_app_event_extras_by_type(
+                app_id_info["v1_name"]
+            ),
         }
 
         super().generate_per_app_id(
@@ -117,6 +129,16 @@ def generate_per_app(
         ):
             return
 
+        extras_by_type = defaultdict(set)
+        for app_id_info in app_ids_info:
+            app_id_extras_by_type = get_glean_app_event_extras_by_type(
+                app_id_info["v1_name"]
+            )
+            for app_id_extra_type, app_id_extras in app_id_extras_by_type.items():
+                extras_by_type[app_id_extra_type].update(app_id_extras)
+
+        custom_render_kwargs = {"extras_by_type": extras_by_type}
+
         super().generate_per_app(
             project_id,
             app_name,
@@ -126,4 +148,72 @@ def generate_per_app(
             parallelism=parallelism,
             id_token=id_token,
             all_base_tables_exist=all_base_tables_exist,
+            custom_render_kwargs=custom_render_kwargs,
         )
+
+
+@cache
+def get_glean_app_repository(v1_name: str) -> dict:
+    """Return the Glean app's repository."""
+    for repository in get_glean_repositories():
+        if repository["name"] == v1_name:
+            return repository
+    raise Exception(f"No Glean repository found for app `{v1_name}`.")
+
+
+@cache
+def get_glean_dependency_v1_name(dependency: str) -> str:
+    """Return the v1 name of the Glean dependency, which may be a library."""
+    for repository in get_glean_repositories():
+        if "library_names" in repository and dependency in repository["library_names"]:
+            return repository["name"]
+    return dependency
+
+
+@cache
+def get_glean_app_event_extras_by_type(
+    v1_name: str, ping: str = "events", cutoff_date: Optional[date] = None
+) -> dict[str, set[str]]:
+    """Return the Glean app's event extra keys for the specified ping, grouped by type."""
+    extras_by_type = defaultdict(set)
+    repository = get_glean_app_repository(v1_name)
+
+    if not cutoff_date:
+        ping_delete_after_days = (
+            (
+                repository.get("moz_pipeline_metadata", {})
+                .get(ping, {})
+                .get("expiration_policy", {})
+                .get("delete_after_days")
+            )
+            or (
+                repository.get("moz_pipeline_metadata_defaults", {})
+                .get("expiration_policy", {})
+                .get("delete_after_days")
+            )
+            or 775
+        )
+        cutoff_date = date.today() - timedelta(days=(ping_delete_after_days - 1))
+
+    metrics = list(get_glean_app_metrics(v1_name).values())
+    for metric in metrics:
+        if metric["type"] == "event":
+            for history in metric["history"]:
+                last_date = datetime.fromisoformat(history["dates"]["last"]).date()
+                if ping in history["send_in_pings"] and last_date >= cutoff_date:
+                    for extra_key, extra_key_info in history["extra_keys"].items():
+                        extra_type = extra_key_info.get("type", "string")
+                        extras_by_type[extra_type].add(extra_key)
+
+    for dependency in repository["dependencies"]:
+        dependency_v1_name = get_glean_dependency_v1_name(dependency)
+        dependency_extras_by_type = get_glean_app_event_extras_by_type(
+            dependency_v1_name, ping, cutoff_date
+        )
+        for (
+            dependency_extra_type,
+            dependency_extras,
+        ) in dependency_extras_by_type.items():
+            extras_by_type[dependency_extra_type].update(dependency_extras)
+
+    return extras_by_type

sql_generators/glean_usage/templates/cross_channel_events_stream.query.sql

@@ -1,18 +1,27 @@
+{% from 'macros.sql' import event_extras_by_type_struct -%}
 -- Generated via ./bqetl generate glean_usage
 CREATE OR REPLACE VIEW
   `{{ project_id }}.{{ target_view }}`
 AS
-{% for (dataset, channel) in datasets %}
-{% if not loop.first -%}
-UNION ALL
-{% endif -%}
+WITH events_stream_union AS (
+  {% for (dataset, channel) in datasets %}
+    {% if not loop.first -%}
+      UNION ALL BY NAME
+    {% endif -%}
+    SELECT
+      "{{ dataset }}" AS normalized_app_id,
+      e.*
+      {% if app_name == "fenix" -%}
+        REPLACE(mozfun.norm.fenix_app_info("{{ dataset }}", client_info.app_build).channel AS normalized_channel),
+      {% elif datasets|length > 1 -%}
+        REPLACE("{{ channel }}" AS normalized_channel),
+      {% endif -%}
+    FROM `{{ project_id }}.{{ dataset }}_derived.events_stream_v1` AS e
+  {% endfor %}
+)
 SELECT
-  "{{ dataset }}" AS normalized_app_id,
-  e.*
-  {% if app_name == "fenix" -%}
-  REPLACE(mozfun.norm.fenix_app_info("{{ dataset }}", client_info.app_build).channel AS normalized_channel),
-  {% elif datasets|length > 1 -%}
-  REPLACE("{{ channel }}" AS normalized_channel),
-  {% endif -%}
-FROM `{{ project_id }}.{{ dataset }}.events_stream` AS e
-{% endfor %}
+  *,
+  {% if extras_by_type %}
+    {{ event_extras_by_type_struct(extras_by_type) }} AS extras
+  {% endif %}
+FROM events_stream_union

sql_generators/glean_usage/templates/events_stream.view.sql

@@ -1,9 +1,13 @@
+{% from 'macros.sql' import event_extras_by_type_struct -%}
 {{ header }}
 
 CREATE OR REPLACE VIEW
   `{{ project_id }}.{{ events_stream_view }}`
 AS
 SELECT
-  *
+  *,
+  {% if extras_by_type %}
+    {{ event_extras_by_type_struct(extras_by_type) }} AS extras
+  {% endif %}
 FROM
   `{{ project_id }}.{{ events_stream_table }}`

sql_generators/glean_usage/templates/macros.sql

@@ -30,3 +30,22 @@ _core_clients_first_seen AS (
     _fennec_id_lookup.client_id
 )
 {% endmacro %}
+
+
+{% macro event_extras_by_type_struct(extras_by_type) %}
+  STRUCT(
+    {% for extra_type in extras_by_type | sort %}
+      STRUCT(
+        {% for extra in extras_by_type[extra_type] | sort %}
+          {% if extra_type == 'boolean' %}
+            LAX_BOOL(event_extra.{{ extra }}) AS `{{ extra }}`{{ ',' if not loop.last else '' }}
+          {% elif extra_type == 'quantity' %}
+            LAX_INT64(event_extra.{{ extra }}) AS `{{ extra }}`{{ ',' if not loop.last else '' }}
+          {% else %}
+            JSON_VALUE(event_extra.{{ extra }}) AS `{{ extra }}`{{ ',' if not loop.last else '' }}
+          {% endif %}
+        {% endfor %}
+      ) AS `{{ extra_type }}`{{ ',' if not loop.last else '' }}
+    {% endfor %}
+  )
+{% endmacro %}