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            "submission_timestamp_granularity",
393            "expiration_policy",
394            "override_attributes",
395            "jwe_mappings",
396        ]
397        reordered_metadata = {
398            k: metadata[k] for k in desired_order_list if k in metadata
399        }
400
401        # re-order jwe-mappings
402        desired_order_list = ["source_field_path", "decrypted_field_path"]
403        jwe_mapping_metadata = reordered_metadata.get("jwe_mappings")
404        if jwe_mapping_metadata:
405            reordered_jwe_mapping_metadata = []
406            for mapping in jwe_mapping_metadata:
407                reordered_jwe_mapping_metadata.append(
408                    {k: mapping[k] for k in desired_order_list if k in mapping}
409                )
410            reordered_metadata["jwe_mappings"] = reordered_jwe_mapping_metadata
411
412        # future proofing, in case there are other fields added at the ping top level
413        # add them to the end.
414        leftovers = {k: metadata[k] for k in set(metadata) - set(reordered_metadata)}
415        reordered_metadata = {**reordered_metadata, **leftovers}
416        return reordered_metadata
417
418    def get_pings_and_pipeline_metadata(self) -> Dict[str, Dict]:
419        pings = self._get_ping_data_and_dependencies_with_default_metadata()
420        for ping_name, ping_data in pings.items():
421            metadata = ping_data.get("moz_pipeline_metadata")
422            if not metadata:
423                continue
424            metadata["include_info_sections"] = self._is_field_included(
425                ping_data, "include_info_sections", consider_all_history=False
426            )
427            metadata["include_client_id"] = self._is_field_included(
428                ping_data, "include_client_id"
429            )
430
431            # While technically unnecessary, the dictionary elements are re-ordered to match the
432            # currently deployed order and used to verify no difference in output.
433            pings[ping_name] = GleanPing.reorder_metadata(metadata)
434        return pings
435
436    def get_ping_descriptions(self) -> Dict[str, str]:
437        return {
438            k: v["history"][-1]["description"] for k, v in self._get_ping_data().items()
439        }
440
441    @staticmethod
442    def _is_field_included(ping_data, field_name, consider_all_history=True) -> bool:
443        """Return false if the field exists and is false.
444
445        If `consider_all_history` is False, then only check the latest value in the ping history.
446
447        Otherwise, if the field is not found or true in one or more history entries,
448        true is returned.
449        """
450
451        # Default to true if not specified.
452        if "history" not in ping_data or len(ping_data["history"]) == 0:
453            return True
454
455        # Check if at some point in the past the field has already been deployed.
456        # And if the caller of this method wants to consider this history of the field.
457        # Keep them in the schema, even if the field has changed as
458        # removing fields is currently not supported.
459        # See https://bugzilla.mozilla.org/show_bug.cgi?id=1898105
460        # and https://bugzilla.mozilla.org/show_bug.cgi?id=1898105#c10
461        ping_history: list
462        if consider_all_history:
463            ping_history = ping_data["history"]
464        else:
465            ping_history = [ping_data["history"][-1]]
466        for history in ping_history:
467            if field_name not in history or history[field_name]:
468                return True
469
470        # The ping was created with include_info_sections = False. The fields can be excluded.
471        return False
472
473    def set_schema_url(self, metadata):
474        """
475        Switch between the glean-min and glean schemas if the ping does not require
476        info sections as specified in the parsed ping info in probe scraper.
477        """
478        if not metadata["include_info_sections"]:
479            self.schema_url = SCHEMA_URL_TEMPLATE.format(
480                branch=self.branch_name
481            ) + SCHEMA_VERSION_TEMPLATE.format(
482                schema_type="glean-min", version=self.version
483            )
484        else:
485            self.schema_url = SCHEMA_URL_TEMPLATE.format(
486                branch=self.branch_name
487            ) + SCHEMA_VERSION_TEMPLATE.format(
488                schema_type="glean", version=self.version
489            )
490
491    def generate_schema(
492        self,
493        config,
494        generic_schema=False,
495    ) -> Dict[str, Schema]:
496        pings = self.get_pings_and_pipeline_metadata()
497        schemas = {}
498
499        for ping, pipeline_meta in pings.items():
500            matchers = {
501                loc: m.clone(new_table_group=ping) for loc, m in config.matchers.items()
502            }
503
504            # Four newly introduced metric types were incorrectly deployed
505            # as repeated key/value structs in all Glean ping tables existing prior
506            # to November 2021. We maintain the incorrect fields for existing tables
507            # by disabling the associated matchers.
508            # Note that each of these types now has a "2" matcher ("text2", "url2", etc.)
509            # defined that will allow metrics of these types to be injected into proper
510            # structs. The gcp-ingestion repository includes logic to rewrite these
511            # metrics under the "2" names.
512            # See https://bugzilla.mozilla.org/show_bug.cgi?id=1737656
513            bq_identifier = "{bq_dataset_family}.{bq_table}".format(**pipeline_meta)
514            if bq_identifier in self.bug_1737656_affected_tables:
515                matchers = {
516                    loc: m
517                    for loc, m in matchers.items()
518                    if not m.matcher.get("bug_1737656_affected")
519                }
520
521            for matcher in matchers.values():
522                matcher.matcher["send_in_pings"]["contains"] = ping
523
524            new_config = Config(ping, matchers=matchers)
525
526            defaults = {"mozPipelineMetadata": pipeline_meta}
527
528            # Adjust the schema path if the ping does not require info sections
529            self.set_schema_url(pipeline_meta)
530            if generic_schema:  # Use the generic glean ping schema
531                schema = self.get_schema(generic_schema=True)
532                schema.schema.update(defaults)
533                schemas[new_config.name] = schema
534            else:
535                generated = super().generate_schema(new_config)
536                for schema in generated.values():
537                    # We want to override each individual key with assembled defaults,
538                    # but keep values _inside_ them if they have been set in the schemas.
539                    for key, value in defaults.items():
540                        if key not in schema.schema:
541                            schema.schema[key] = {}
542                        schema.schema[key].update(value)
543                schemas.update(generated)
544
545        return schemas
546
547    @staticmethod
548    def get_repos():
549        """
550        Retrieve metadata for all non-library Glean repositories
551        """
552        repos = GleanPing._get_json(GleanPing.repos_url)
553        return [repo for repo in repos if "library_names" not in repo]
554
555    def get_app_name(self) -> str:
556        """Get app name associated with the app id.
557
558        e.g. org-mozilla-firefox -> fenix
559        """
560        apps = GleanPing._get_json(GleanPing.app_listings_url)
561        # app id in app-listings has "." instead of "-" so using document_namespace
562        app_name = [
563            app["app_name"] for app in apps if app["document_namespace"] == self.app_id
564        ]
565        return app_name[0] if len(app_name) > 0 else self.app_id
566
567    @staticmethod
568    def get_metric_blocklist():
569        with open(METRIC_BLOCKLIST, "r") as f:
570            return yaml.safe_load(f)