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 logging
  8from pathlib import Path
  9from typing import Dict, List, Set
 10
 11from requests import HTTPError
 12
 13from .config import Config
 14from .generic_ping import GenericPing
 15from .probes import GleanProbe
 16from .schema import Schema
 17
 18ROOT_DIR = Path(__file__).parent
 19BUG_1737656_TXT = ROOT_DIR / "configs" / "bug_1737656_affected.txt"
 20
 21logger = logging.getLogger(__name__)
 22
 23DEFAULT_SCHEMA_URL = (
 24    "https://raw.githubusercontent.com"
 25    "/mozilla-services/mozilla-pipeline-schemas"
 26    "/{branch}/schemas/glean/glean/glean.1.schema.json"
 27)
 28
 29MINIMUM_SCHEMA_URL = (
 30    "https://raw.githubusercontent.com"
 31    "/mozilla-services/mozilla-pipeline-schemas"
 32    "/{branch}/schemas/glean/glean/glean-min.1.schema.json"
 33)
 34
 35
 36class GleanPing(GenericPing):
 37    probes_url_template = GenericPing.probe_info_base_url + "/glean/{}/metrics"
 38    ping_url_template = GenericPing.probe_info_base_url + "/glean/{}/pings"
 39    repos_url = GenericPing.probe_info_base_url + "/glean/repositories"
 40    dependencies_url_template = (
 41        GenericPing.probe_info_base_url + "/glean/{}/dependencies"
 42    )
 43
 44    default_dependencies = ["glean-core"]
 45
 46    with open(BUG_1737656_TXT, "r") as f:
 47        bug_1737656_affected_tables = [
 48            line.strip() for line in f.readlines() if line.strip()
 49        ]
 50
 51    def __init__(self, repo, **kwargs):  # TODO: Make env-url optional
 52        self.repo = repo
 53        self.repo_name = repo["name"]
 54        self.app_id = repo["app_id"]
 55        super().__init__(
 56            DEFAULT_SCHEMA_URL,
 57            DEFAULT_SCHEMA_URL,
 58            self.probes_url_template.format(self.repo_name),
 59            **kwargs,
 60        )
 61
 62    def get_schema(self, generic_schema=False) -> Schema:
 63        """
 64        Fetch schema via URL.
 65
 66        Unless *generic_schema* is set to true, this function makes some modifications
 67        to allow some workarounds for proper injection of metrics.
 68        """
 69        schema = super().get_schema()
 70        if generic_schema:
 71            return schema
 72
 73        # We need to inject placeholders for the url2, text2, etc. types as part
 74        # of mitigation for https://bugzilla.mozilla.org/show_bug.cgi?id=1737656
 75        for metric_name in ["labeled_rate", "jwe", "url", "text"]:
 76            metric1 = schema.get(
 77                ("properties", "metrics", "properties", metric_name)
 78            ).copy()
 79            metric1 = schema.set_schema_elem(
 80                ("properties", "metrics", "properties", metric_name + "2"),
 81                metric1,
 82            )
 83
 84        return schema
 85
 86    def get_dependencies(self):
 87        # Get all of the library dependencies for the application that
 88        # are also known about in the repositories file.
 89
 90        # The dependencies are specified using library names, but we need to
 91        # map those back to the name of the repository in the repository file.
 92        try:
 93            dependencies = self._get_json(
 94                self.dependencies_url_template.format(self.repo_name)
 95            )
 96        except HTTPError:
 97            logging.info(f"For {self.repo_name}, using default Glean dependencies")
 98            return self.default_dependencies
 99
100        dependency_library_names = list(dependencies.keys())
101
102        repos = GleanPing._get_json(GleanPing.repos_url)
103        repos_by_dependency_name = {}
104        for repo in repos:
105            for library_name in repo.get("library_names", []):
106                repos_by_dependency_name[library_name] = repo["name"]
107
108        dependencies = []
109        for name in dependency_library_names:
110            if name in repos_by_dependency_name:
111                dependencies.append(repos_by_dependency_name[name])
112
113        if len(dependencies) == 0:
114            logging.info(f"For {self.repo_name}, using default Glean dependencies")
115            return self.default_dependencies
116
117        logging.info(f"For {self.repo_name}, found Glean dependencies: {dependencies}")
118        return dependencies
119
120    def get_probes(self) -> List[GleanProbe]:
121        data = self._get_json(self.probes_url)
122        probes = list(data.items())
123
124        for dependency in self.get_dependencies():
125            dependency_probes = self._get_json(
126                self.probes_url_template.format(dependency)
127            )
128            probes += list(dependency_probes.items())
129
130        pings = self.get_pings()
131
132        processed = []
133        for _id, defn in probes:
134            probe = GleanProbe(_id, defn, pings=pings)
135            processed.append(probe)
136
137            # Manual handling of incompatible schema changes
138            issue_118_affected = {
139                "fenix",
140                "fenix-nightly",
141                "firefox-android-nightly",
142                "firefox-android-beta",
143                "firefox-android-release",
144            }
145            if (
146                self.repo_name in issue_118_affected
147                and probe.get_name() == "installation.timestamp"
148            ):
149                logging.info(f"Writing column {probe.get_name()} for compatibility.")
150                # See: https://github.com/mozilla/mozilla-schema-generator/issues/118
151                # Search through history for the "string" type and add a copy of
152                # the probe at that time in history. The changepoint signifies
153                # this event.
154                changepoint_index = 0
155                for definition in probe.definition_history:
156                    if definition["type"] != probe.get_type():
157                        break
158                    changepoint_index += 1
159                # Modify the definition with the truncated history.
160                hist_defn = defn.copy()
161                hist_defn[probe.history_key] = probe.definition_history[
162                    changepoint_index:
163                ]
164                hist_defn["type"] = hist_defn[probe.history_key][0]["type"]
165                incompatible_probe_type = GleanProbe(_id, hist_defn, pings=pings)
166                processed.append(incompatible_probe_type)
167
168            # Handling probe type changes (Bug 1870317)
169            probe_types = {hist["type"] for hist in defn[probe.history_key]}
170            if len(probe_types) > 1:
171                # The probe type changed at some point in history.
172                # Create schema entry for each type.
173                hist_defn = defn.copy()
174
175                # No new entry needs to be created for the current probe type
176                probe_types.remove(defn["type"])
177
178                for hist in hist_defn[probe.history_key]:
179                    # Create a new entry for a historic type
180                    if hist["type"] in probe_types:
181                        hist_defn["type"] = hist["type"]
182                        probe = GleanProbe(_id, hist_defn, pings=pings)
183                        processed.append(probe)
184
185                        # Keep track of the types entries were already created for
186                        probe_types.remove(hist["type"])
187
188        return processed
189
190    def _get_ping_data(self) -> Dict[str, Dict]:
191        url = self.ping_url_template.format(self.repo_name)
192        ping_data = GleanPing._get_json(url)
193        for dependency in self.get_dependencies():
194            dependency_pings = self._get_json(self.ping_url_template.format(dependency))
195            ping_data.update(dependency_pings)
196        return ping_data
197
198    def _get_ping_data_without_dependencies(self) -> Dict[str, Dict]:
199        url = self.ping_url_template.format(self.repo_name)
200        ping_data = GleanPing._get_json(url)
201        return ping_data
202
203    def _get_dependency_pings(self, dependency):
204        return self._get_json(self.ping_url_template.format(dependency))
205
206    def get_pings(self) -> Set[str]:
207        return self._get_ping_data().keys()
208
209    @staticmethod
210    def apply_default_metadata(ping_metadata, default_metadata):
211        """apply_default_metadata recurses down into dicts nested
212        to an arbitrary depth, updating keys. The ``default_metadata`` is merged into
213        ``ping_metadata``.
214        :param ping_metadata: dict onto which the merge is executed
215        :param default_metadata: dct merged into ping_metadata
216        :return: None
217        """
218        for k, v in default_metadata.items():
219            if (
220                k in ping_metadata
221                and isinstance(ping_metadata[k], dict)
222                and isinstance(default_metadata[k], dict)
223            ):
224                GleanPing.apply_default_metadata(ping_metadata[k], default_metadata[k])
225            else:
226                ping_metadata[k] = default_metadata[k]
227
228    def _get_ping_data_and_dependencies_with_default_metadata(self) -> Dict[str, Dict]:
229        # Get the ping data with the pipeline metadata
230        ping_data = self._get_ping_data_without_dependencies()
231
232        # The ping endpoint for the dependency pings does not include any repo defined
233        # moz_pipeline_metadata_defaults so they need to be applied here.
234
235        # 1.  Get repo and pipeline default metadata.
236        repos = self.get_repos()
237        current_repo = next((x for x in repos if x.get("app_id") == self.app_id), {})
238        default_metadata = current_repo.get("moz_pipeline_metadata_defaults", {})
239
240        # 2.  Apply the default metadata to each dependency defined ping.
241
242        # Apply app-level metadata to pings defined in dependencies
243        app_metadata = current_repo.get("moz_pipeline_metadata", {})
244
245        for dependency in self.get_dependencies():
246            dependency_pings = self._get_dependency_pings(dependency)
247            for dependency_ping in dependency_pings.values():
248                # Although it is counter intuitive to apply the default metadata on top of the
249                # existing dependency ping metadata it does set the repo specific value for
250                # bq_dataset_family instead of using the dependency id for the bq_dataset_family
251                # value.
252                GleanPing.apply_default_metadata(
253                    dependency_ping.get("moz_pipeline_metadata"), default_metadata
254                )
255                # app-level ping properties take priority over the app defaults
256                metadata_override = app_metadata.get(dependency_ping["name"])
257                if metadata_override is not None:
258                    GleanPing.apply_default_metadata(
259                        dependency_ping.get("moz_pipeline_metadata"), metadata_override
260                    )
261            ping_data.update(dependency_pings)
262
263        return ping_data
264
265    @staticmethod
266    def reorder_metadata(metadata):
267        desired_order_list = [
268            "bq_dataset_family",
269            "bq_table",
270            "bq_metadata_format",
271            "include_info_sections",
272            "submission_timestamp_granularity",
273            "expiration_policy",
274            "override_attributes",
275            "jwe_mappings",
276        ]
277        reordered_metadata = {
278            k: metadata[k] for k in desired_order_list if k in metadata
279        }
280
281        # re-order jwe-mappings
282        desired_order_list = ["source_field_path", "decrypted_field_path"]
283        jwe_mapping_metadata = reordered_metadata.get("jwe_mappings")
284        if jwe_mapping_metadata:
285            reordered_jwe_mapping_metadata = []
286            for mapping in jwe_mapping_metadata:
287                reordered_jwe_mapping_metadata.append(
288                    {k: mapping[k] for k in desired_order_list if k in mapping}
289                )
290            reordered_metadata["jwe_mappings"] = reordered_jwe_mapping_metadata
291
292        # future proofing, in case there are other fields added at the ping top level
293        # add them to the end.
294        leftovers = {k: metadata[k] for k in set(metadata) - set(reordered_metadata)}
295        reordered_metadata = {**reordered_metadata, **leftovers}
296        return reordered_metadata
297
298    def get_pings_and_pipeline_metadata(self) -> Dict[str, Dict]:
299        pings = self._get_ping_data_and_dependencies_with_default_metadata()
300        for ping_name, ping_data in pings.items():
301            metadata = ping_data.get("moz_pipeline_metadata")
302            if not metadata:
303                continue
304            metadata["include_info_sections"] = self._is_field_included(
305                ping_data, "include_info_sections", consider_all_history=False
306            )
307            metadata["include_client_id"] = self._is_field_included(
308                ping_data, "include_client_id"
309            )
310
311            # While technically unnecessary, the dictionary elements are re-ordered to match the
312            # currently deployed order and used to verify no difference in output.
313            pings[ping_name] = GleanPing.reorder_metadata(metadata)
314        return pings
315
316    def get_ping_descriptions(self) -> Dict[str, str]:
317        return {
318            k: v["history"][-1]["description"] for k, v in self._get_ping_data().items()
319        }
320
321    @staticmethod
322    def _is_field_included(ping_data, field_name, consider_all_history=True) -> bool:
323        """Return false if the field exists and is false.
324
325        If `consider_all_history` is False, then only check the latest value in the ping history.
326
327        Otherwise, if the field is not found or true in one or more history entries,
328        true is returned.
329        """
330
331        # Default to true if not specified.
332        if "history" not in ping_data or len(ping_data["history"]) == 0:
333            return True
334
335        # Check if at some point in the past the field has already been deployed.
336        # And if the caller of this method wants to consider this history of the field.
337        # Keep them in the schema, even if the field has changed as
338        # removing fields is currently not supported.
339        # See https://bugzilla.mozilla.org/show_bug.cgi?id=1898105
340        # and https://bugzilla.mozilla.org/show_bug.cgi?id=1898105#c10
341        ping_history: list
342        if consider_all_history:
343            ping_history = ping_data["history"]
344        else:
345            ping_history = [ping_data["history"][-1]]
346        for history in ping_history:
347            if field_name not in history or history[field_name]:
348                return True
349
350        # The ping was created with include_info_sections = False. The fields can be excluded.
351        return False
352
353    def set_schema_url(self, metadata):
354        """
355        Switch between the glean-min and glean schemas if the ping does not require
356        info sections as specified in the parsed ping info in probe scraper.
357        """
358        if not metadata["include_info_sections"]:
359            self.schema_url = MINIMUM_SCHEMA_URL.format(branch=self.branch_name)
360        else:
361            self.schema_url = DEFAULT_SCHEMA_URL.format(branch=self.branch_name)
362
363    def generate_schema(self, config, generic_schema=False) -> Dict[str, Schema]:
364        pings = self.get_pings_and_pipeline_metadata()
365        schemas = {}
366
367        for ping, pipeline_meta in pings.items():
368            matchers = {
369                loc: m.clone(new_table_group=ping) for loc, m in config.matchers.items()
370            }
371
372            # Four newly introduced metric types were incorrectly deployed
373            # as repeated key/value structs in all Glean ping tables existing prior
374            # to November 2021. We maintain the incorrect fields for existing tables
375            # by disabling the associated matchers.
376            # Note that each of these types now has a "2" matcher ("text2", "url2", etc.)
377            # defined that will allow metrics of these types to be injected into proper
378            # structs. The gcp-ingestion repository includes logic to rewrite these
379            # metrics under the "2" names.
380            # See https://bugzilla.mozilla.org/show_bug.cgi?id=1737656
381            bq_identifier = "{bq_dataset_family}.{bq_table}".format(**pipeline_meta)
382            if bq_identifier in self.bug_1737656_affected_tables:
383                matchers = {
384                    loc: m
385                    for loc, m in matchers.items()
386                    if not m.matcher.get("bug_1737656_affected")
387                }
388
389            for matcher in matchers.values():
390                matcher.matcher["send_in_pings"]["contains"] = ping
391            new_config = Config(ping, matchers=matchers)
392
393            defaults = {"mozPipelineMetadata": pipeline_meta}
394
395            # Adjust the schema path if the ping does not require info sections
396            self.set_schema_url(pipeline_meta)
397            if generic_schema:  # Use the generic glean ping schema
398                schema = self.get_schema(generic_schema=True)
399                schema.schema.update(defaults)
400                schemas[new_config.name] = schema
401            else:
402                generated = super().generate_schema(new_config)
403                for schema in generated.values():
404                    # We want to override each individual key with assembled defaults,
405                    # but keep values _inside_ them if they have been set in the schemas.
406                    for key, value in defaults.items():
407                        if key not in schema.schema:
408                            schema.schema[key] = {}
409                        schema.schema[key].update(value)
410                schemas.update(generated)
411
412        return schemas
413
414    @staticmethod
415    def get_repos():
416        """
417        Retrieve metadata for all non-library Glean repositories
418        """
419        repos = GleanPing._get_json(GleanPing.repos_url)
420        return [repo for repo in repos if "library_names" not in repo]