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
 23
 24class GleanPing(GenericPing):
 25
 26    schema_url = (
 27        "https://raw.githubusercontent.com/mozilla-services/mozilla-pipeline-schemas"
 28        "/{branch}/schemas/glean/glean/glean.1.schema.json"
 29    )
 30    probes_url_template = GenericPing.probe_info_base_url + "/glean/{}/metrics"
 31    ping_url_template = GenericPing.probe_info_base_url + "/glean/{}/pings"
 32    repos_url = GenericPing.probe_info_base_url + "/glean/repositories"
 33    dependencies_url_template = (
 34        GenericPing.probe_info_base_url + "/glean/{}/dependencies"
 35    )
 36
 37    default_dependencies = ["glean-core"]
 38    ignore_pings = {
 39        "all-pings",
 40        "all_pings",
 41        "default",
 42        "glean_ping_info",
 43        "glean_client_info",
 44    }
 45
 46    with open(BUG_1737656_TXT, "r") as f:
 47        bug_1737656_affected_tables = [l.strip() for l in f.readlines() if l.strip()]
 48
 49    def __init__(self, repo, **kwargs):  # TODO: Make env-url optional
 50        self.repo = repo
 51        self.repo_name = repo["name"]
 52        self.app_id = repo["app_id"]
 53        super().__init__(
 54            self.schema_url,
 55            self.schema_url,
 56            self.probes_url_template.format(self.repo_name),
 57            **kwargs,
 58        )
 59
 60    def get_schema(self, generic_schema=False) -> Schema:
 61        """
 62        Fetch schema via URL.
 63
 64        Unless *generic_schema* is set to true, this function makes some modifications
 65        to allow some workarounds for proper injection of metrics.
 66        """
 67        schema = super().get_schema()
 68        if generic_schema:
 69            return schema
 70
 71        # We need to inject placeholders for the url2, text2, etc. types as part
 72        # of mitigation for https://bugzilla.mozilla.org/show_bug.cgi?id=1737656
 73        for metric_name in ["labeled_rate", "jwe", "url", "text"]:
 74            metric1 = schema.get(
 75                ("properties", "metrics", "properties", metric_name)
 76            ).copy()
 77            metric1 = schema.set_schema_elem(
 78                ("properties", "metrics", "properties", metric_name + "2"),
 79                metric1,
 80            )
 81
 82        return schema
 83
 84    def get_dependencies(self):
 85        # Get all of the library dependencies for the application that
 86        # are also known about in the repositories file.
 87
 88        # The dependencies are specified using library names, but we need to
 89        # map those back to the name of the repository in the repository file.
 90        try:
 91            dependencies = self._get_json(
 92                self.dependencies_url_template.format(self.repo_name)
 93            )
 94        except HTTPError:
 95            logging.info(f"For {self.repo_name}, using default Glean dependencies")
 96            return self.default_dependencies
 97
 98        dependency_library_names = list(dependencies.keys())
 99
100        repos = GleanPing._get_json(GleanPing.repos_url)
101        repos_by_dependency_name = {}
102        for repo in repos:
103            for library_name in repo.get("library_names", []):
104                repos_by_dependency_name[library_name] = repo["name"]
105
106        dependencies = []
107        for name in dependency_library_names:
108            if name in repos_by_dependency_name:
109                dependencies.append(repos_by_dependency_name[name])
110
111        if len(dependencies) == 0:
112            logging.info(f"For {self.repo_name}, using default Glean dependencies")
113            return self.default_dependencies
114
115        logging.info(f"For {self.repo_name}, found Glean dependencies: {dependencies}")
116        return dependencies
117
118    def get_probes(self) -> List[GleanProbe]:
119        data = self._get_json(self.probes_url)
120        probes = list(data.items())
121
122        for dependency in self.get_dependencies():
123            dependency_probes = self._get_json(
124                self.probes_url_template.format(dependency)
125            )
126            probes += list(dependency_probes.items())
127
128        pings = self.get_pings()
129
130        processed = []
131        for _id, defn in probes:
132            probe = GleanProbe(_id, defn, pings=pings)
133            processed.append(probe)
134
135            # Manual handling of incompatible schema changes
136            issue_118_affected = {
137                "fenix",
138                "fenix-nightly",
139                "firefox-android-nightly",
140                "firefox-android-beta",
141                "firefox-android-release",
142            }
143            if (
144                self.repo_name in issue_118_affected
145                and probe.get_name() == "installation.timestamp"
146            ):
147                logging.info(f"Writing column {probe.get_name()} for compatibility.")
148                # See: https://github.com/mozilla/mozilla-schema-generator/issues/118
149                # Search through history for the "string" type and add a copy of
150                # the probe at that time in history. The changepoint signifies
151                # this event.
152                changepoint_index = 0
153                for definition in probe.definition_history:
154                    if definition["type"] != probe.get_type():
155                        break
156                    changepoint_index += 1
157                # Modify the definition with the truncated history.
158                hist_defn = defn.copy()
159                hist_defn[probe.history_key] = probe.definition_history[
160                    changepoint_index:
161                ]
162                hist_defn["type"] = hist_defn[probe.history_key][0]["type"]
163                incompatible_probe_type = GleanProbe(_id, hist_defn, pings=pings)
164                processed.append(incompatible_probe_type)
165
166        return processed
167
168    def _get_ping_data(self) -> Dict[str, Dict]:
169        url = self.ping_url_template.format(self.repo_name)
170        ping_data = GleanPing._get_json(url)
171        for dependency in self.get_dependencies():
172            dependency_pings = self._get_json(self.ping_url_template.format(dependency))
173            ping_data.update(dependency_pings)
174        return ping_data
175
176    def _get_ping_data_without_dependencies(self) -> Dict[str, Dict]:
177        url = self.ping_url_template.format(self.repo_name)
178        ping_data = GleanPing._get_json(url)
179        return ping_data
180
181    def _get_dependency_pings(self, dependency):
182        return self._get_json(self.ping_url_template.format(dependency))
183
184    def get_pings(self) -> Set[str]:
185        return self._get_ping_data().keys()
186
187    @staticmethod
188    def apply_default_metadata(ping_metadata, default_metadata):
189        """apply_default_metadata recurses down into dicts nested
190        to an arbitrary depth, updating keys. The ``default_metadata`` is merged into
191        ``ping_metadata``.
192        :param ping_metadata: dict onto which the merge is executed
193        :param default_metadata: dct merged into ping_metadata
194        :return: None
195        """
196        for k, v in default_metadata.items():
197            if (
198                k in ping_metadata
199                and isinstance(ping_metadata[k], dict)
200                and isinstance(default_metadata[k], dict)
201            ):
202                GleanPing.apply_default_metadata(ping_metadata[k], default_metadata[k])
203            else:
204                ping_metadata[k] = default_metadata[k]
205
206    def _get_ping_data_and_dependencies_with_default_metadata(self) -> Dict[str, Dict]:
207        # Get the ping data with the pipeline metadata
208        ping_data = self._get_ping_data_without_dependencies()
209
210        # The ping endpoint for the dependency pings does not include any repo defined
211        # moz_pipeline_metadata_defaults so they need to be applied here.
212
213        # 1.  Get repo and pipeline default metadata.
214        repos = GleanPing.get_repos()
215        current_repo = next((x for x in repos if x.get("app_id") == self.app_id), {})
216        default_metadata = current_repo.get("moz_pipeline_metadata_defaults", {})
217
218        # 2.  Apply the default metadata to each dependency defined ping.
219        for dependency in self.get_dependencies():
220            dependency_pings = self._get_dependency_pings(dependency)
221            for dependency_ping in dependency_pings.values():
222                # Although it is counter intuitive to apply the default metadata on top of the
223                # existing dependency ping metadata it does set the repo specific value for
224                # bq_dataset_family instead of using the dependency id for the bq_dataset_family
225                # value.
226                GleanPing.apply_default_metadata(
227                    dependency_ping.get("moz_pipeline_metadata"), default_metadata
228                )
229            ping_data.update(dependency_pings)
230        return ping_data
231
232    @staticmethod
233    def reorder_metadata(metadata):
234        desired_order_list = [
235            "bq_dataset_family",
236            "bq_table",
237            "bq_metadata_format",
238            "submission_timestamp_granularity",
239            "expiration_policy",
240            "override_attributes",
241            "jwe_mappings",
242        ]
243        reordered_metadata = {
244            k: metadata[k] for k in desired_order_list if k in metadata
245        }
246
247        # re-order jwe-mappings
248        desired_order_list = ["source_field_path", "decrypted_field_path"]
249        jwe_mapping_metadata = reordered_metadata.get("jwe_mappings")
250        if jwe_mapping_metadata:
251            reordered_jwe_mapping_metadata = []
252            for mapping in jwe_mapping_metadata:
253                reordered_jwe_mapping_metadata.append(
254                    {k: mapping[k] for k in desired_order_list if k in mapping}
255                )
256            reordered_metadata["jwe_mappings"] = reordered_jwe_mapping_metadata
257
258        # future proofing, in case there are other fields added at the ping top level
259        # add them to the end.
260        leftovers = {k: metadata[k] for k in set(metadata) - set(reordered_metadata)}
261        reordered_metadata = {**reordered_metadata, **leftovers}
262        return reordered_metadata
263
264    def get_pings_and_pipeline_metadata(self) -> Dict[str, Dict]:
265        pings = self._get_ping_data_and_dependencies_with_default_metadata()
266        for ping_name, ping_data in pings.items():
267            metadata = ping_data.get("moz_pipeline_metadata")
268
269            # While technically unnecessary, the dictionary elements are re-ordered to match the
270            # currently deployed order and used to verify no difference in output.
271            pings[ping_name] = GleanPing.reorder_metadata(metadata)
272        return pings
273
274    def get_ping_descriptions(self) -> Dict[str, str]:
275        return {
276            k: v["history"][-1]["description"] for k, v in self._get_ping_data().items()
277        }
278
279    def generate_schema(
280        self, config, split, generic_schema=False
281    ) -> Dict[str, List[Schema]]:
282        pings = self.get_pings_and_pipeline_metadata()
283        schemas = {}
284
285        for ping, pipeline_meta in pings.items():
286            matchers = {
287                loc: m.clone(new_table_group=ping) for loc, m in config.matchers.items()
288            }
289
290            # Four newly introduced metric types were incorrectly deployed
291            # as repeated key/value structs in all Glean ping tables existing prior
292            # to November 2021. We maintain the incorrect fields for existing tables
293            # by disabling the associated matchers.
294            # Note that each of these types now has a "2" matcher ("text2", "url2", etc.)
295            # defined that will allow metrics of these types to be injected into proper
296            # structs. The gcp-ingestion repository includes logic to rewrite these
297            # metrics under the "2" names.
298            # See https://bugzilla.mozilla.org/show_bug.cgi?id=1737656
299            bq_identifier = "{bq_dataset_family}.{bq_table}".format(**pipeline_meta)
300            if bq_identifier in self.bug_1737656_affected_tables:
301                matchers = {
302                    loc: m
303                    for loc, m in matchers.items()
304                    if not m.matcher.get("bug_1737656_affected")
305                }
306
307            for matcher in matchers.values():
308                matcher.matcher["send_in_pings"]["contains"] = ping
309            new_config = Config(ping, matchers=matchers)
310
311            defaults = {"mozPipelineMetadata": pipeline_meta}
312
313            if generic_schema:  # Use the generic glean ping schema
314                schema = self.get_schema(generic_schema=True)
315                schema.schema.update(defaults)
316                schemas[new_config.name] = [schema]
317            else:
318                generated = super().generate_schema(new_config)
319                for value in generated.values():
320                    for schema in value:
321                        schema.schema.update(defaults)
322                schemas.update(generated)
323
324        return schemas
325
326    @staticmethod
327    def get_repos():
328        """
329        Retrieve metadata for all non-library Glean repositories
330        """
331        repos = GleanPing._get_json(GleanPing.repos_url)
332        return [repo for repo in repos if "library_names" not in repo]