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