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
 42class GleanPing(GenericPing):
 43    probes_url_template = GenericPing.probe_info_base_url + "/glean/{}/metrics"
 44    ping_url_template = GenericPing.probe_info_base_url + "/glean/{}/pings"
 45    repos_url = GenericPing.probe_info_base_url + "/glean/repositories"
 46    dependencies_url_template = (
 47        GenericPing.probe_info_base_url + "/glean/{}/dependencies"
 48    )
 49    app_listings_url = GenericPing.probe_info_base_url + "/v2/glean/app-listings"
 50
 51    default_dependencies = ["glean-core"]
 52
 53    with open(BUG_1737656_TXT, "r") as f:
 54        bug_1737656_affected_tables = [
 55            line.strip() for line in f.readlines() if line.strip()
 56        ]
 57
 58    def __init__(
 59        self, repo, version=1, use_metrics_blocklist=False, **kwargs
 60    ):  # TODO: Make env-url optional
 61        self.repo = repo
 62        self.repo_name = repo["name"]
 63        self.app_id = repo["app_id"]
 64        self.version = version
 65
 66        if use_metrics_blocklist:
 67            self.metric_blocklist = self.get_metric_blocklist()
 68        else:
 69            self.metric_blocklist = {}
 70
 71        super().__init__(
 72            DEFAULT_SCHEMA_URL,
 73            DEFAULT_SCHEMA_URL,
 74            self.probes_url_template.format(self.repo_name),
 75            **kwargs,
 76        )
 77
 78    def get_schema(self, generic_schema=False) -> Schema:
 79        """
 80        Fetch schema via URL.
 81
 82        Unless *generic_schema* is set to true, this function makes some modifications
 83        to allow some workarounds for proper injection of metrics.
 84        """
 85        schema = super().get_schema()
 86        if generic_schema:
 87            return schema
 88
 89        # We need to inject placeholders for the url2, text2, etc. types as part
 90        # of mitigation for https://bugzilla.mozilla.org/show_bug.cgi?id=1737656
 91        for metric_name in ["labeled_rate", "jwe", "url", "text"]:
 92            metric1 = schema.get(
 93                ("properties", "metrics", "properties", metric_name)
 94            ).copy()
 95            metric1 = schema.set_schema_elem(
 96                ("properties", "metrics", "properties", metric_name + "2"),
 97                metric1,
 98            )
 99
100        return schema
101
102    @cache
103    def get_dependencies(self):
104        # Get all of the library dependencies for the application that
105        # are also known about in the repositories file.
106
107        # The dependencies are specified using library names, but we need to
108        # map those back to the name of the repository in the repository file.
109        try:
110            dependencies = self._get_json(
111                self.dependencies_url_template.format(self.repo_name)
112            )
113        except HTTPError:
114            logging.info(f"For {self.repo_name}, using default Glean dependencies")
115            return self.default_dependencies
116
117        dependency_library_names = list(dependencies.keys())
118
119        repos = GleanPing._get_json(GleanPing.repos_url)
120        repos_by_dependency_name = {}
121        for repo in repos:
122            for library_name in repo.get("library_names", []):
123                repos_by_dependency_name[library_name] = repo["name"]
124
125        dependencies = []
126        for name in dependency_library_names:
127            if name in repos_by_dependency_name:
128                dependencies.append(repos_by_dependency_name[name])
129
130        if len(dependencies) == 0:
131            logging.info(f"For {self.repo_name}, using default Glean dependencies")
132            return self.default_dependencies
133
134        logging.info(f"For {self.repo_name}, found Glean dependencies: {dependencies}")
135        return dependencies
136
137    @staticmethod
138    def remove_pings_from_metric(
139        metric: Dict[str, Any], blocked_pings: List[str]
140    ) -> Dict[str, Any]:
141        """Remove the given pings from the metric's `send_in_pings` history.
142
143        Only removes if the given metric has been removed from the source since a fixed date
144        (2025-01-01). This allows metrics to be added back to the schema.
145        """
146        if (
147            metric["in-source"]
148            or len(blocked_pings) == 0
149            or datetime.fromisoformat(metric["history"][-1]["dates"]["last"])
150            >= datetime(year=2025, month=1, day=1)
151        ):
152            return metric
153
154        for history_entry in metric["history"]:
155            history_entry["send_in_pings"] = [
156                p for p in history_entry["send_in_pings"] if p not in blocked_pings
157            ]
158
159        return metric
160
161    def get_probes(self) -> List[GleanProbe]:
162        data = self._get_json(self.probes_url)
163
164        # blocklist needs to be applied here instead of generate_schema because it needs to be
165        # dependency-aware; metrics can move between app and library and still be in the schema
166        # turn blocklist into metric_name -> ping_types map
167        blocklist = defaultdict(list)
168        for ping_type, metric_names in self.metric_blocklist.get(
169            self.get_app_name(), {}
170        ).items():
171            for metric_name in metric_names:
172                blocklist[metric_name].append(ping_type)
173
174        probes = [
175            (name, self.remove_pings_from_metric(defn, blocklist.get(name, [])))
176            for name, defn in data.items()
177        ]
178
179        for dependency in self.get_dependencies():
180            dependency_probes = self._get_json(
181                self.probes_url_template.format(dependency)
182            )
183
184            dependency_blocklist = defaultdict(list)
185            for ping_type, metric_names in self.metric_blocklist.get(
186                dependency, {}
187            ).items():
188                for metric_name in metric_names:
189                    dependency_blocklist[metric_name].append(ping_type)
190
191            probes += [
192                (
193                    name,
194                    self.remove_pings_from_metric(
195                        defn, dependency_blocklist.get(name, [])
196                    ),
197                )
198                for name, defn in dependency_probes.items()
199            ]
200
201        pings = self.get_pings()
202
203        processed = []
204        for _id, defn in probes:
205            probe = GleanProbe(_id, defn, pings=pings)
206            processed.append(probe)
207
208            # Handling probe type changes (Bug 1870317)
209            probe_types = {hist["type"] for hist in defn[probe.history_key]}
210            if len(probe_types) > 1:
211                # The probe type changed at some point in history.
212                # Create schema entry for each type.
213                hist_defn = defn.copy()
214
215                # No new entry needs to be created for the current probe type
216                probe_types.remove(defn["type"])
217
218                for hist in hist_defn[probe.history_key]:
219                    # Create a new entry for a historic type
220                    if hist["type"] in probe_types:
221                        hist_defn["type"] = hist["type"]
222                        probe = GleanProbe(_id, hist_defn, pings=pings)
223                        processed.append(probe)
224
225                        # Keep track of the types entries were already created for
226                        probe_types.remove(hist["type"])
227
228        return processed
229
230    def _get_ping_data(self) -> Dict[str, Dict]:
231        url = self.ping_url_template.format(self.repo_name)
232        ping_data = GleanPing._get_json(url)
233        for dependency in self.get_dependencies():
234            dependency_pings = self._get_json(self.ping_url_template.format(dependency))
235            ping_data.update(dependency_pings)
236        return ping_data
237
238    def _get_ping_data_without_dependencies(self) -> Dict[str, Dict]:
239        url = self.ping_url_template.format(self.repo_name)
240        ping_data = GleanPing._get_json(url)
241        return ping_data
242
243    def _get_dependency_pings(self, dependency):
244        return self._get_json(self.ping_url_template.format(dependency))
245
246    def get_pings(self) -> Set[str]:
247        return self._get_ping_data().keys()
248
249    @staticmethod
250    def apply_default_metadata(ping_metadata, default_metadata):
251        """apply_default_metadata recurses down into dicts nested
252        to an arbitrary depth, updating keys. The ``default_metadata`` is merged into
253        ``ping_metadata``.
254        :param ping_metadata: dict onto which the merge is executed
255        :param default_metadata: dct merged into ping_metadata
256        :return: None
257        """
258        for k, v in default_metadata.items():
259            if (
260                k in ping_metadata
261                and isinstance(ping_metadata[k], dict)
262                and isinstance(default_metadata[k], dict)
263            ):
264                GleanPing.apply_default_metadata(ping_metadata[k], default_metadata[k])
265            else:
266                ping_metadata[k] = default_metadata[k]
267
268    def _get_ping_data_and_dependencies_with_default_metadata(self) -> Dict[str, Dict]:
269        # Get the ping data with the pipeline metadata
270        ping_data = self._get_ping_data_without_dependencies()
271
272        # The ping endpoint for the dependency pings does not include any repo defined
273        # moz_pipeline_metadata_defaults so they need to be applied here.
274
275        # 1.  Get repo and pipeline default metadata.
276        repos = self.get_repos()
277        current_repo = next((x for x in repos if x.get("app_id") == self.app_id), {})
278        default_metadata = current_repo.get("moz_pipeline_metadata_defaults", {})
279
280        # 2.  Apply the default metadata to each dependency defined ping.
281
282        # Apply app-level metadata to pings defined in dependencies
283        app_metadata = current_repo.get("moz_pipeline_metadata", {})
284
285        for dependency in self.get_dependencies():
286            dependency_pings = self._get_dependency_pings(dependency)
287            for dependency_ping in dependency_pings.values():
288                # Although it is counter intuitive to apply the default metadata on top of the
289                # existing dependency ping metadata it does set the repo specific value for
290                # bq_dataset_family instead of using the dependency id for the bq_dataset_family
291                # value.
292                GleanPing.apply_default_metadata(
293                    dependency_ping.get("moz_pipeline_metadata"),
294                    copy.deepcopy(default_metadata),
295                )
296                # app-level ping properties take priority over the app defaults
297                metadata_override = app_metadata.get(dependency_ping["name"])
298                if metadata_override is not None:
299                    GleanPing.apply_default_metadata(
300                        dependency_ping.get("moz_pipeline_metadata"), metadata_override
301                    )
302            ping_data.update(dependency_pings)
303
304        return ping_data
305
306    @staticmethod
307    def reorder_metadata(metadata):
308        desired_order_list = [
309            "bq_dataset_family",
310            "bq_table",
311            "bq_metadata_format",
312            "include_info_sections",
313            "submission_timestamp_granularity",
314            "expiration_policy",
315            "override_attributes",
316            "jwe_mappings",
317        ]
318        reordered_metadata = {
319            k: metadata[k] for k in desired_order_list if k in metadata
320        }
321
322        # re-order jwe-mappings
323        desired_order_list = ["source_field_path", "decrypted_field_path"]
324        jwe_mapping_metadata = reordered_metadata.get("jwe_mappings")
325        if jwe_mapping_metadata:
326            reordered_jwe_mapping_metadata = []
327            for mapping in jwe_mapping_metadata:
328                reordered_jwe_mapping_metadata.append(
329                    {k: mapping[k] for k in desired_order_list if k in mapping}
330                )
331            reordered_metadata["jwe_mappings"] = reordered_jwe_mapping_metadata
332
333        # future proofing, in case there are other fields added at the ping top level
334        # add them to the end.
335        leftovers = {k: metadata[k] for k in set(metadata) - set(reordered_metadata)}
336        reordered_metadata = {**reordered_metadata, **leftovers}
337        return reordered_metadata
338
339    def get_pings_and_pipeline_metadata(self) -> Dict[str, Dict]:
340        pings = self._get_ping_data_and_dependencies_with_default_metadata()
341        for ping_name, ping_data in pings.items():
342            metadata = ping_data.get("moz_pipeline_metadata")
343            if not metadata:
344                continue
345            metadata["include_info_sections"] = self._is_field_included(
346                ping_data, "include_info_sections", consider_all_history=False
347            )
348            metadata["include_client_id"] = self._is_field_included(
349                ping_data, "include_client_id"
350            )
351
352            # While technically unnecessary, the dictionary elements are re-ordered to match the
353            # currently deployed order and used to verify no difference in output.
354            pings[ping_name] = GleanPing.reorder_metadata(metadata)
355        return pings
356
357    def get_ping_descriptions(self) -> Dict[str, str]:
358        return {
359            k: v["history"][-1]["description"] for k, v in self._get_ping_data().items()
360        }
361
362    @staticmethod
363    def _is_field_included(ping_data, field_name, consider_all_history=True) -> bool:
364        """Return false if the field exists and is false.
365
366        If `consider_all_history` is False, then only check the latest value in the ping history.
367
368        Otherwise, if the field is not found or true in one or more history entries,
369        true is returned.
370        """
371
372        # Default to true if not specified.
373        if "history" not in ping_data or len(ping_data["history"]) == 0:
374            return True
375
376        # Check if at some point in the past the field has already been deployed.
377        # And if the caller of this method wants to consider this history of the field.
378        # Keep them in the schema, even if the field has changed as
379        # removing fields is currently not supported.
380        # See https://bugzilla.mozilla.org/show_bug.cgi?id=1898105
381        # and https://bugzilla.mozilla.org/show_bug.cgi?id=1898105#c10
382        ping_history: list
383        if consider_all_history:
384            ping_history = ping_data["history"]
385        else:
386            ping_history = [ping_data["history"][-1]]
387        for history in ping_history:
388            if field_name not in history or history[field_name]:
389                return True
390
391        # The ping was created with include_info_sections = False. The fields can be excluded.
392        return False
393
394    def set_schema_url(self, metadata):
395        """
396        Switch between the glean-min and glean schemas if the ping does not require
397        info sections as specified in the parsed ping info in probe scraper.
398        """
399        if not metadata["include_info_sections"]:
400            self.schema_url = SCHEMA_URL_TEMPLATE.format(
401                branch=self.branch_name
402            ) + SCHEMA_VERSION_TEMPLATE.format(
403                schema_type="glean-min", version=self.version
404            )
405        else:
406            self.schema_url = SCHEMA_URL_TEMPLATE.format(
407                branch=self.branch_name
408            ) + SCHEMA_VERSION_TEMPLATE.format(
409                schema_type="glean", version=self.version
410            )
411
412    def generate_schema(
413        self,
414        config,
415        generic_schema=False,
416        blocked_distribution_pings=("events", "baseline"),
417    ) -> Dict[str, Schema]:
418        pings = self.get_pings_and_pipeline_metadata()
419        schemas = {}
420
421        for ping, pipeline_meta in pings.items():
422            matchers = {
423                loc: m.clone(new_table_group=ping) for loc, m in config.matchers.items()
424            }
425
426            # Four newly introduced metric types were incorrectly deployed
427            # as repeated key/value structs in all Glean ping tables existing prior
428            # to November 2021. We maintain the incorrect fields for existing tables
429            # by disabling the associated matchers.
430            # Note that each of these types now has a "2" matcher ("text2", "url2", etc.)
431            # defined that will allow metrics of these types to be injected into proper
432            # structs. The gcp-ingestion repository includes logic to rewrite these
433            # metrics under the "2" names.
434            # See https://bugzilla.mozilla.org/show_bug.cgi?id=1737656
435            bq_identifier = "{bq_dataset_family}.{bq_table}".format(**pipeline_meta)
436            if bq_identifier in self.bug_1737656_affected_tables:
437                matchers = {
438                    loc: m
439                    for loc, m in matchers.items()
440                    if not m.matcher.get("bug_1737656_affected")
441                }
442
443            for matcher in matchers.values():
444                matcher.matcher["send_in_pings"]["contains"] = ping
445
446                # temporarily block distributions from being added to events and baseline pings
447                # https://mozilla-hub.atlassian.net/browse/DENG-10606
448                if (
449                    blocked_distribution_pings
450                    and ping in blocked_distribution_pings
451                    and matcher.type.endswith("_distribution")
452                ):
453                    matcher.matcher["send_in_pings"]["not_contains"] = ping
454
455            new_config = Config(ping, matchers=matchers)
456
457            defaults = {"mozPipelineMetadata": pipeline_meta}
458
459            # Adjust the schema path if the ping does not require info sections
460            self.set_schema_url(pipeline_meta)
461            if generic_schema:  # Use the generic glean ping schema
462                schema = self.get_schema(generic_schema=True)
463                schema.schema.update(defaults)
464                schemas[new_config.name] = schema
465            else:
466                generated = super().generate_schema(new_config)
467                for schema in generated.values():
468                    # We want to override each individual key with assembled defaults,
469                    # but keep values _inside_ them if they have been set in the schemas.
470                    for key, value in defaults.items():
471                        if key not in schema.schema:
472                            schema.schema[key] = {}
473                        schema.schema[key].update(value)
474                schemas.update(generated)
475
476        return schemas
477
478    @staticmethod
479    def get_repos():
480        """
481        Retrieve metadata for all non-library Glean repositories
482        """
483        repos = GleanPing._get_json(GleanPing.repos_url)
484        return [repo for repo in repos if "library_names" not in repo]
485
486    def get_app_name(self) -> str:
487        """Get app name associated with the app id.
488
489        e.g. org-mozilla-firefox -> fenix
490        """
491        apps = GleanPing._get_json(GleanPing.app_listings_url)
492        # app id in app-listings has "." instead of "-" so using document_namespace
493        app_name = [
494            app["app_name"] for app in apps if app["document_namespace"] == self.app_id
495        ]
496        return app_name[0] if len(app_name) > 0 else self.app_id
497
498    @staticmethod
499    def get_metric_blocklist():
500        with open(METRIC_BLOCKLIST, "r") as f:
501            return yaml.safe_load(f)