mozilla_schema_generator.glean_ping

  1# -*- coding: utf-8 -*-
  2
  3# This Source Code Form is subject to the terms of the Mozilla Public
  4# License, v. 2.0. If a copy of the MPL was not distributed with this
  5# file, You can obtain one at http://mozilla.org/MPL/2.0/.
  6
  7import copy
  8import logging
  9from collections import defaultdict
 10from datetime import datetime
 11from functools import cache
 12from pathlib import Path
 13from typing import Any, Dict, List, Set
 14
 15import yaml
 16from requests import HTTPError
 17
 18from .config import Config
 19from .generic_ping import GenericPing
 20from .probes import GleanProbe
 21from .schema import Schema
 22
 23ROOT_DIR = Path(__file__).parent
 24BUG_1737656_TXT = ROOT_DIR / "configs" / "bug_1737656_affected.txt"
 25METRIC_BLOCKLIST = ROOT_DIR / "configs" / "metric_blocklist.yaml"
 26
 27logger = logging.getLogger(__name__)
 28
 29SCHEMA_URL_TEMPLATE = (
 30    "https://raw.githubusercontent.com"
 31    "/mozilla-services/mozilla-pipeline-schemas"
 32    "/{branch}/schemas/glean/glean/"
 33)
 34
 35SCHEMA_VERSION_TEMPLATE = "{schema_type}.{version}.schema.json"
 36
 37DEFAULT_SCHEMA_URL = SCHEMA_URL_TEMPLATE + SCHEMA_VERSION_TEMPLATE.format(
 38    schema_type="glean", version=1
 39)
 40
 41# App ids with v2 schemas already deployed. Applying the metric blocklist to these would drop
 42# columns that already exist in their tables, which isn't allowed, so they keep every metric.
 43METRIC_BLOCKLIST_EXEMPT_APP_IDS = frozenset(
 44    {
 45        "org-mozilla-fenix-nightly",
 46        "org-mozilla-fennec-aurora",
 47    }
 48)
 49
 50
 51class GleanPing(GenericPing):
 52    probes_url_template = GenericPing.probe_info_base_url + "/glean/{}/metrics"
 53    ping_url_template = GenericPing.probe_info_base_url + "/glean/{}/pings"
 54    repos_url = GenericPing.probe_info_base_url + "/glean/repositories"
 55    dependencies_url_template = (
 56        GenericPing.probe_info_base_url + "/glean/{}/dependencies"
 57    )
 58    app_listings_url = GenericPing.probe_info_base_url + "/v2/glean/app-listings"
 59
 60    default_dependencies = ["glean-core"]
 61
 62    with open(BUG_1737656_TXT, "r") as f:
 63        bug_1737656_affected_tables = [
 64            line.strip() for line in f.readlines() if line.strip()
 65        ]
 66
 67    def __init__(
 68        self, repo, version=1, use_metrics_blocklist=False, **kwargs
 69    ):  # TODO: Make env-url optional
 70        self.repo = repo
 71        self.repo_name = repo["name"]
 72        self.app_id = repo["app_id"]
 73        self.version = version
 74
 75        if use_metrics_blocklist and self.app_id not in METRIC_BLOCKLIST_EXEMPT_APP_IDS:
 76            self.metric_blocklist = self.get_metric_blocklist()
 77        else:
 78            self.metric_blocklist = {}
 79
 80        super().__init__(
 81            DEFAULT_SCHEMA_URL,
 82            DEFAULT_SCHEMA_URL,
 83            self.probes_url_template.format(self.repo_name),
 84            **kwargs,
 85        )
 86
 87    def get_schema(self, generic_schema=False) -> Schema:
 88        """
 89        Fetch schema via URL.
 90
 91        Unless *generic_schema* is set to true, this function makes some modifications
 92        to allow some workarounds for proper injection of metrics.
 93        """
 94        schema = super().get_schema()
 95        if generic_schema:
 96            return schema
 97
 98        # We need to inject placeholders for the url2, text2, etc. types as part
 99        # of mitigation for https://bugzilla.mozilla.org/show_bug.cgi?id=1737656
100        for metric_name in ["labeled_rate", "jwe", "url", "text"]:
101            metric1 = schema.get(
102                ("properties", "metrics", "properties", metric_name)
103            ).copy()
104            metric1 = schema.set_schema_elem(
105                ("properties", "metrics", "properties", metric_name + "2"),
106                metric1,
107            )
108
109        return schema
110
111    @cache
112    def get_dependencies(self):
113        # Get all of the library dependencies for the application that
114        # are also known about in the repositories file.
115
116        # The dependencies are specified using library names, but we need to
117        # map those back to the name of the repository in the repository file.
118        try:
119            dependencies = self._get_json(
120                self.dependencies_url_template.format(self.repo_name)
121            )
122        except HTTPError:
123            logging.info(f"For {self.repo_name}, using default Glean dependencies")
124            return self.default_dependencies
125
126        dependency_library_names = list(dependencies.keys())
127
128        repos = GleanPing._get_json(GleanPing.repos_url)
129        repos_by_dependency_name = {}
130        for repo in repos:
131            for library_name in repo.get("library_names", []):
132                repos_by_dependency_name[library_name] = repo["name"]
133
134        dependencies = []
135        for name in dependency_library_names:
136            if name in repos_by_dependency_name:
137                dependencies.append(repos_by_dependency_name[name])
138
139        if len(dependencies) == 0:
140            logging.info(f"For {self.repo_name}, using default Glean dependencies")
141            return self.default_dependencies
142
143        logging.info(f"For {self.repo_name}, found Glean dependencies: {dependencies}")
144        return dependencies
145
146    @staticmethod
147    def remove_pings_from_metric(
148        metric: Dict[str, Any], blocked_pings: List[str]
149    ) -> Dict[str, Any]:
150        """Remove the given pings from the metric's `send_in_pings` history.
151
152        Only removes if the given metric has been removed from the source since a fixed date
153        (2025-01-01). This allows metrics to be added back to the schema.
154        """
155        if (
156            metric["in-source"]
157            or len(blocked_pings) == 0
158            or datetime.fromisoformat(metric["history"][-1]["dates"]["last"])
159            >= datetime(year=2025, month=8, day=1)
160        ):
161            return metric
162
163        for history_entry in metric["history"]:
164            history_entry["send_in_pings"] = [
165                p for p in history_entry["send_in_pings"] if p not in blocked_pings
166            ]
167
168        return metric
169
170    def get_probes(self) -> List[GleanProbe]:
171        data = self._get_json(self.probes_url)
172
173        # blocklist needs to be applied here instead of generate_schema because it needs to be
174        # dependency-aware; metrics can move between app and library and still be in the schema
175        # turn blocklist into metric_name -> ping_types map
176        blocklist = defaultdict(list)
177        for ping_type, metric_names in self.metric_blocklist.get(
178            self.get_app_name(), {}
179        ).items():
180            for metric_name in metric_names:
181                blocklist[metric_name].append(ping_type)
182
183        probes = [
184            (name, self.remove_pings_from_metric(defn, blocklist.get(name, [])))
185            for name, defn in data.items()
186        ]
187
188        for dependency in self.get_dependencies():
189            dependency_probes = self._get_json(
190                self.probes_url_template.format(dependency)
191            )
192
193            dependency_blocklist = defaultdict(list)
194            for ping_type, metric_names in self.metric_blocklist.get(
195                dependency, {}
196            ).items():
197                for metric_name in metric_names:
198                    dependency_blocklist[metric_name].append(ping_type)
199
200            probes += [
201                (
202                    name,
203                    self.remove_pings_from_metric(
204                        defn, dependency_blocklist.get(name, [])
205                    ),
206                )
207                for name, defn in dependency_probes.items()
208            ]
209
210        # A metric can be moved between an app and its dependencies or between dependencies while
211        # probe scraper keeps the history in each location, so both definitions are returned
212        # Merge the history per probe to take the latest definition while still being able to
213        # find metric type changes below
214        # Metrics are not merged if they are not sent in the same pings as they are disjoint
215
216        # Metrics are grouped by their normalized BigQuery column name (from jsonschema-transpiler)
217        # rather than their raw name. e.g. "media.audio.init_failure" and "media.audio_init_failure"
218        # normalize to "media_audio_init_failure". The transpiler picks one of the colliding
219        # descriptions non-deterministically.
220        def _normalize_name(name):
221            return name.replace(".", "_").replace("-", "_")
222
223        def _pings_in_history(defn):
224            return {
225                p
226                for h in defn[GleanProbe.history_key]
227                for p in h.get("send_in_pings", ["metrics"])
228            }
229
230        def _latest_history_date(defn):
231            return max(
232                datetime.fromisoformat(h["dates"]["last"])
233                for h in defn[GleanProbe.history_key]
234            )
235
236        def _dedupe_sort_key(defn):
237            """Prefer the most recent definition, breaking ties by choosing the in-source metric."""
238            return (
239                _latest_history_date(defn),
240                defn.get(GleanProbe.in_source_key, False),
241                defn["name"],
242            )
243
244        # Group probes that share a normalized name and whose pings intersect to combine
245        # moved metrics and metrics that only differ by "." vs "_"
246        grouped_by_name: Dict[str, List[List[dict]]] = defaultdict(list)
247        for name, defn in probes:
248            defn_pings = _pings_in_history(defn)
249            existing_groups = grouped_by_name[_normalize_name(name)]
250            matches = [
251                group
252                for group in existing_groups
253                if any(
254                    _pings_in_history(other_defn) & defn_pings for other_defn in group
255                )
256            ]
257            if not matches:
258                existing_groups.append([defn])
259            else:
260                merged_group = [defn]
261                for g in matches:
262                    merged_group.extend(g)
263                    existing_groups.remove(g)
264                existing_groups.append(merged_group)
265
266        # Take latest definition per group
267        deduped_probes: List[Any] = []
268        for groups in grouped_by_name.values():
269            for group in groups:
270                latest_defn = max(group, key=_dedupe_sort_key)
271                if len(group) > 1:
272                    latest_defn = latest_defn.copy()
273                    latest_defn[GleanProbe.history_key] = sorted(
274                        (h for d in group for h in d[GleanProbe.history_key]),
275                        key=lambda h: datetime.fromisoformat(h["dates"]["first"]),
276                    )
277                deduped_probes.append((latest_defn["name"], latest_defn))
278        probes = deduped_probes
279
280        pings = self.get_pings()
281
282        processed = []
283        for _id, defn in probes:
284            probe = GleanProbe(_id, defn, pings=pings)
285            processed.append(probe)
286
287            # Handling probe type changes (Bug 1870317)
288            probe_types = {hist["type"] for hist in defn[probe.history_key]}
289            if len(probe_types) > 1:
290                # The probe type changed at some point in history.
291                # Create schema entry for each type.
292                hist_defn = defn.copy()
293
294                # No new entry needs to be created for the current probe type
295                probe_types.remove(defn["type"])
296
297                for hist in hist_defn[probe.history_key]:
298                    # Create a new entry for a historic type
299                    if hist["type"] in probe_types:
300                        hist_defn["type"] = hist["type"]
301                        probe = GleanProbe(_id, hist_defn, pings=pings)
302                        processed.append(probe)
303
304                        # Keep track of the types entries were already created for
305                        probe_types.remove(hist["type"])
306
307        return processed
308
309    def _get_ping_data(self) -> Dict[str, Dict]:
310        url = self.ping_url_template.format(self.repo_name)
311        ping_data = GleanPing._get_json(url)
312        for dependency in self.get_dependencies():
313            dependency_pings = self._get_json(self.ping_url_template.format(dependency))
314            ping_data.update(dependency_pings)
315        return ping_data
316
317    def _get_ping_data_without_dependencies(self) -> Dict[str, Dict]:
318        url = self.ping_url_template.format(self.repo_name)
319        ping_data = GleanPing._get_json(url)
320        return ping_data
321
322    def _get_dependency_pings(self, dependency):
323        return self._get_json(self.ping_url_template.format(dependency))
324
325    def get_pings(self) -> Set[str]:
326        return self._get_ping_data().keys()
327
328    @staticmethod
329    def apply_default_metadata(ping_metadata, default_metadata):
330        """apply_default_metadata recurses down into dicts nested
331        to an arbitrary depth, updating keys. The ``default_metadata`` is merged into
332        ``ping_metadata``.
333        :param ping_metadata: dict onto which the merge is executed
334        :param default_metadata: dct merged into ping_metadata
335        :return: None
336        """
337        for k, v in default_metadata.items():
338            if (
339                k in ping_metadata
340                and isinstance(ping_metadata[k], dict)
341                and isinstance(default_metadata[k], dict)
342            ):
343                GleanPing.apply_default_metadata(ping_metadata[k], default_metadata[k])
344            else:
345                ping_metadata[k] = default_metadata[k]
346
347    def _get_ping_data_and_dependencies_with_default_metadata(self) -> Dict[str, Dict]:
348        # Get the ping data with the pipeline metadata
349        ping_data = self._get_ping_data_without_dependencies()
350
351        # The ping endpoint for the dependency pings does not include any repo defined
352        # moz_pipeline_metadata_defaults so they need to be applied here.
353
354        # 1.  Get repo and pipeline default metadata.
355        repos = self.get_repos()
356        current_repo = next((x for x in repos if x.get("app_id") == self.app_id), {})
357        default_metadata = current_repo.get("moz_pipeline_metadata_defaults", {})
358
359        # 2.  Apply the default metadata to each dependency defined ping.
360
361        # Apply app-level metadata to pings defined in dependencies
362        app_metadata = current_repo.get("moz_pipeline_metadata", {})
363
364        for dependency in self.get_dependencies():
365            dependency_pings = self._get_dependency_pings(dependency)
366            for dependency_ping in dependency_pings.values():
367                # Although it is counter intuitive to apply the default metadata on top of the
368                # existing dependency ping metadata it does set the repo specific value for
369                # bq_dataset_family instead of using the dependency id for the bq_dataset_family
370                # value.
371                GleanPing.apply_default_metadata(
372                    dependency_ping.get("moz_pipeline_metadata"),
373                    copy.deepcopy(default_metadata),
374                )
375                # app-level ping properties take priority over the app defaults
376                metadata_override = app_metadata.get(dependency_ping["name"])
377                if metadata_override is not None:
378                    GleanPing.apply_default_metadata(
379                        dependency_ping.get("moz_pipeline_metadata"), metadata_override
380                    )
381            ping_data.update(dependency_pings)
382
383        return ping_data
384
385    @staticmethod
386    def reorder_metadata(metadata):
387        desired_order_list = [
388            "bq_dataset_family",
389            "bq_table",
390            "bq_metadata_format",
391            "include_info_sections",
392            "uploader_capabilities",
393            "submission_timestamp_granularity",
394            "expiration_policy",
395            "override_attributes",
396            "jwe_mappings",
397        ]
398        reordered_metadata = {
399            k: metadata[k] for k in desired_order_list if k in metadata
400        }
401
402        # re-order jwe-mappings
403        desired_order_list = ["source_field_path", "decrypted_field_path"]
404        jwe_mapping_metadata = reordered_metadata.get("jwe_mappings")
405        if jwe_mapping_metadata:
406            reordered_jwe_mapping_metadata = []
407            for mapping in jwe_mapping_metadata:
408                reordered_jwe_mapping_metadata.append(
409                    {k: mapping[k] for k in desired_order_list if k in mapping}
410                )
411            reordered_metadata["jwe_mappings"] = reordered_jwe_mapping_metadata
412
413        # future proofing, in case there are other fields added at the ping top level
414        # add them to the end.
415        leftovers = {k: metadata[k] for k in set(metadata) - set(reordered_metadata)}
416        reordered_metadata = {**reordered_metadata, **leftovers}
417        return reordered_metadata
418
419    def get_pings_and_pipeline_metadata(self) -> Dict[str, Dict]:
420        pings = self._get_ping_data_and_dependencies_with_default_metadata()
421        for ping_name, ping_data in pings.items():
422            metadata = ping_data.get("moz_pipeline_metadata")
423            if not metadata:
424                continue
425            metadata["include_info_sections"] = self._is_field_included(
426                ping_data, "include_info_sections", consider_all_history=False
427            )
428            metadata["include_client_id"] = self._is_field_included(
429                ping_data, "include_client_id"
430            )
431            # Omitted when empty to avoid touching the schemas of every ping without
432            # uploader capabilities.
433            uploader_capabilities = self._get_uploader_capabilities(ping_data)
434            if uploader_capabilities:
435                metadata["uploader_capabilities"] = uploader_capabilities
436
437            # While technically unnecessary, the dictionary elements are re-ordered to match the
438            # currently deployed order and used to verify no difference in output.
439            pings[ping_name] = GleanPing.reorder_metadata(metadata)
440        return pings
441
442    @staticmethod
443    def _get_uploader_capabilities(ping_data) -> List[str]:
444        """Return the uploader capabilities from the latest history entry, e.g. ["ohttp"]."""
445        history = ping_data.get("history") or [{}]
446        return history[-1].get("uploader_capabilities") or []
447
448    def get_ohttp_pings_with_info_sections(self) -> List[str]:
449        """Return pings that declare the `ohttp` uploader capability but still
450        include info sections.
451        """
452        pings = self._get_ping_data_and_dependencies_with_default_metadata()
453        return sorted(
454            ping_name
455            for ping_name, ping_data in pings.items()
456            # pings without pipeline metadata don't get a schema
457            if ping_data.get("moz_pipeline_metadata")
458            and "ohttp" in self._get_uploader_capabilities(ping_data)
459            and self._is_field_included(
460                ping_data, "include_info_sections", consider_all_history=False
461            )
462        )
463
464    def get_ping_descriptions(self) -> Dict[str, str]:
465        return {
466            k: v["history"][-1]["description"] for k, v in self._get_ping_data().items()
467        }
468
469    @staticmethod
470    def _is_field_included(ping_data, field_name, consider_all_history=True) -> bool:
471        """Return false if the field exists and is false.
472
473        If `consider_all_history` is False, then only check the latest value in the ping history.
474
475        Otherwise, if the field is not found or true in one or more history entries,
476        true is returned.
477        """
478
479        # Default to true if not specified.
480        if "history" not in ping_data or len(ping_data["history"]) == 0:
481            return True
482
483        # Check if at some point in the past the field has already been deployed.
484        # And if the caller of this method wants to consider this history of the field.
485        # Keep them in the schema, even if the field has changed as
486        # removing fields is currently not supported.
487        # See https://bugzilla.mozilla.org/show_bug.cgi?id=1898105
488        # and https://bugzilla.mozilla.org/show_bug.cgi?id=1898105#c10
489        ping_history: list
490        if consider_all_history:
491            ping_history = ping_data["history"]
492        else:
493            ping_history = [ping_data["history"][-1]]
494        for history in ping_history:
495            if field_name not in history or history[field_name]:
496                return True
497
498        # The ping was created with include_info_sections = False. The fields can be excluded.
499        return False
500
501    def set_schema_url(self, metadata):
502        """
503        Switch between the glean-min and glean schemas if the ping does not require
504        info sections as specified in the parsed ping info in probe scraper.
505        """
506        if not metadata["include_info_sections"]:
507            self.schema_url = SCHEMA_URL_TEMPLATE.format(
508                branch=self.branch_name
509            ) + SCHEMA_VERSION_TEMPLATE.format(
510                schema_type="glean-min", version=self.version
511            )
512        else:
513            self.schema_url = SCHEMA_URL_TEMPLATE.format(
514                branch=self.branch_name
515            ) + SCHEMA_VERSION_TEMPLATE.format(
516                schema_type="glean", version=self.version
517            )
518
519    def generate_schema(
520        self,
521        config,
522        generic_schema=False,
523    ) -> Dict[str, Schema]:
524        pings = self.get_pings_and_pipeline_metadata()
525        schemas = {}
526
527        for ping, pipeline_meta in pings.items():
528            matchers = {
529                loc: m.clone(new_table_group=ping) for loc, m in config.matchers.items()
530            }
531
532            # Four newly introduced metric types were incorrectly deployed
533            # as repeated key/value structs in all Glean ping tables existing prior
534            # to November 2021. We maintain the incorrect fields for existing tables
535            # by disabling the associated matchers.
536            # Note that each of these types now has a "2" matcher ("text2", "url2", etc.)
537            # defined that will allow metrics of these types to be injected into proper
538            # structs. The gcp-ingestion repository includes logic to rewrite these
539            # metrics under the "2" names.
540            # See https://bugzilla.mozilla.org/show_bug.cgi?id=1737656
541            bq_identifier = "{bq_dataset_family}.{bq_table}".format(**pipeline_meta)
542            if bq_identifier in self.bug_1737656_affected_tables:
543                matchers = {
544                    loc: m
545                    for loc, m in matchers.items()
546                    if not m.matcher.get("bug_1737656_affected")
547                }
548
549            for matcher in matchers.values():
550                matcher.matcher["send_in_pings"]["contains"] = ping
551
552            new_config = Config(ping, matchers=matchers)
553
554            defaults = {"mozPipelineMetadata": pipeline_meta}
555
556            # Adjust the schema path if the ping does not require info sections
557            self.set_schema_url(pipeline_meta)
558            if generic_schema:  # Use the generic glean ping schema
559                schema = self.get_schema(generic_schema=True)
560                schema.schema.update(defaults)
561                schemas[new_config.name] = schema
562            else:
563                generated = super().generate_schema(new_config)
564                for schema in generated.values():
565                    # We want to override each individual key with assembled defaults,
566                    # but keep values _inside_ them if they have been set in the schemas.
567                    for key, value in defaults.items():
568                        if key not in schema.schema:
569                            schema.schema[key] = {}
570                        schema.schema[key].update(value)
571                schemas.update(generated)
572
573        return schemas
574
575    @staticmethod
576    def get_repos():
577        """
578        Retrieve metadata for all non-library Glean repositories
579        """
580        repos = GleanPing._get_json(GleanPing.repos_url)
581        return [repo for repo in repos if "library_names" not in repo]
582
583    def get_app_name(self) -> str:
584        """Get app name associated with the app id.
585
586        e.g. org-mozilla-firefox -> fenix
587        """
588        apps = GleanPing._get_json(GleanPing.app_listings_url)
589        # app id in app-listings has "." instead of "-" so using document_namespace
590        app_name = [
591            app["app_name"] for app in apps if app["document_namespace"] == self.app_id
592        ]
593        return app_name[0] if len(app_name) > 0 else self.app_id
594
595    @staticmethod
596    def get_metric_blocklist():
597        with open(METRIC_BLOCKLIST, "r") as f:
598            return yaml.safe_load(f)