Bug 2077133 - Parse and validate flows.yaml and emit flow registration data

CHANGELOG.md

@@ -1,5 +1,9 @@
 # Changelog
 
+## Unreleased
+
+- Support `flows.yaml` files (`moz://mozilla.org/schemas/glean/flows/1-0-0`), declaring multi-step tasks measured as a unit over existing event metrics. Flows are validated against the metrics passed to the parser, with warnings for events that aren't defined in any of them (`UNRESOLVED_FLOW_EVENT`) and for flows with no exit events (`EMPTY_ENDS_ON`). Kotlin, Swift, and Rust output gains the flow definitions to pass to Glean at initialization (`GleanFlows`), and Markdown output documents flows and the flows each event belongs to. Flows are passed to external translators as the `flows` option ([bug 2077133](https://bugzilla.mozilla.org/show_bug.cgi?id=2077133))
+
 ## 22.0.0
 
 - JS/TS server: Add support for `server-deletion-request` ping. Server apps can now add `server-deletion-request` to `send_in_pings` on identifier metrics (e.g. user IDs) ([DENG-4079](https://mozilla-hub.atlassian.net/browse/DENG-4079))

docs/flows-yaml.rst

@@ -0,0 +1,13 @@
+``flows.yaml`` file
+======================
+
+Documentation for the ``flows.yaml`` file is in the `flows YAML registry format in the Glean user documentation <https://mozilla.github.io/glean/book/reference/yaml/flows.html>`_.
+
+JSON Schema
+-----------
+
+There is a formal schema for validating ``flows.yaml`` files, included in its
+entirety below:
+
+.. literalinclude:: ../glean_parser/schemas/flows.1-0-0.schema.yaml
+   :language: yaml

docs/index.rst

@@ -15,6 +15,7 @@ using Glean is available in `the Glean book <https://mozilla.github.io/glean>`_.
    metrics-yaml
    pings-yaml
    tags-yaml
+   flows-yaml
    modules
    contributing
    authors

glean_parser/__main__.py

@@ -233,6 +233,7 @@ def dump(input, allow_reserved, allow_missing_files, require_tags):
     metrics = {
         metric.identifier(): metric.serialize()
         for category, probes in results.value.items()
+        if not glean_parser.flows.flows_in({category: probes})
         for probe_name, metric in probes.items()
     }
 

glean_parser/data_review.py

@@ -13,6 +13,7 @@
 import re
 
 
+from . import flows
 from . import parser
 from . import util
 
@@ -47,7 +48,9 @@ def generate(
     durations = set()
     responsible_emails = set()
     filtered_metrics = list()
-    for metrics in all_objects.value.values():
+    for category_key, metrics in all_objects.value.items():
+        if flows.flows_in({category_key: metrics}):
+            continue
         for metric in metrics.values():
             if not any([len(reobj.findall(bug)) == 1 for bug in metric.bugs]):
                 continue

glean_parser/flows.py

@@ -0,0 +1,177 @@
+# -*- coding: utf-8 -*-
+
+# This Source Code Form is subject to the terms of the Mozilla Public
+# License, v. 2.0. If a copy of the MPL was not distributed with this
+# file, You can obtain one at http://mozilla.org/MPL/2.0/.
+
+"""
+Classes for each of the high-level flow types.
+"""
+
+from typing import Any, Dict, List, Mapping, Optional, Set
+
+from . import util
+
+
+# Milliseconds per `auto_close_timeout` unit.
+TIMEOUT_UNIT_MS = {
+    "s": 1000,
+    "m": 60 * 1000,
+    "h": 60 * 60 * 1000,
+    "d": 24 * 60 * 60 * 1000,
+}
+
+# The roles an event can play in a flow, in the order they are listed in
+# `flows.yaml`.
+ROLES = ("starts_on", "ends_on", "includes")
+
+
+class Flow:
+    def __init__(
+        self,
+        name: str,
+        description: str,
+        bugs: List[str],
+        data_reviews: List[str],
+        notification_emails: List[str],
+        starts_on: List[str],
+        ends_on: Optional[List[str]] = None,
+        includes: Optional[List[str]] = None,
+        auto_close_timeout: Optional[str] = None,
+        disabled: bool = False,
+        defined_in: Optional[Dict] = None,
+        no_lint: Optional[List[str]] = None,
+        _validated: bool = False,
+    ):
+        # Avoid cyclical import
+        from . import parser
+
+        self.name = name
+        self.description = description
+        self.bugs = bugs
+        self.data_reviews = data_reviews
+        self.notification_emails = notification_emails
+        self.starts_on = starts_on
+        self.ends_on = ends_on if ends_on is not None else []
+        self.includes = includes if includes is not None else []
+        self.auto_close_timeout = auto_close_timeout
+        self.disabled = disabled
+        self.defined_in = defined_in
+        if no_lint is None:
+            no_lint = []
+        self.no_lint = no_lint
+
+        # The events this flow names that are not defined in any of the files
+        # given to the parser, and are assumed to be defined elsewhere. Filled
+        # in once every file has been parsed.
+        self.unresolved: Set[str] = set()
+
+        # _validated indicates whether this flow has already been jsonschema
+        # validated (but not any of the Python-level validation).
+        if not _validated:
+            data: Dict[str, util.JSONType] = {
+                "$schema": parser.FLOWS_ID,
+                self.name: self._serialize_input(),
+            }
+            for error in parser.validate(data):
+                raise ValueError(error)
+
+        if not self.starts_on:
+            raise ValueError("A flow must have at least one event in `starts_on`.")
+
+        # An event may play at most one role in a flow.
+        seen: Dict[str, str] = {}
+        for role in ROLES:
+            for event in getattr(self, role):
+                if event in seen:
+                    raise ValueError(
+                        f"Event '{event}' is in both `{seen[event]}` and `{role}`. "
+                        "An event may appear in at most one of a flow's lists."
+                    )
+                seen[event] = role
+
+    @property
+    def type(self) -> str:
+        return "flow"
+
+    @property
+    def auto_close_timeout_ms(self) -> Optional[int]:
+        """
+        The `auto_close_timeout`, in milliseconds, or `None` if unset.
+        """
+        if self.auto_close_timeout is None:
+            return None
+        value, unit = self.auto_close_timeout[:-1], self.auto_close_timeout[-1]
+        return int(value) * TIMEOUT_UNIT_MS[unit]
+
+    def events(self) -> Dict[str, str]:
+        """
+        Every event this flow names, mapped to its role: `starts_on`,
+        `ends_on`, or `includes`.
+        """
+        return {event: role for role in ROLES for event in getattr(self, role)}
+
+    def _serialize_input(self) -> Dict[str, util.JSONType]:
+        d = self.serialize()
+        modified_dict = util.remove_output_params(d, "defined_in")
+        modified_dict = util.remove_output_params(modified_dict, "unresolved")
+        return modified_dict
+
+    def serialize(self) -> Dict[str, util.JSONType]:
+        """
+        Serialize the flow back to JSON object model.
+        """
+        d = self.__dict__.copy()
+        del d["name"]
+        d["unresolved"] = sorted(self.unresolved)
+        if d["auto_close_timeout"] is None:
+            del d["auto_close_timeout"]
+        return d
+
+
+def flows_in(objs: Mapping[str, Any]) -> Dict[str, "Flow"]:
+    """
+    Returns the flows in a parsed object tree, keyed by name.
+
+    Flows are stored under the `flows` key, which is also a valid metric
+    category name. The key holds flows only if it holds `Flow` objects; the
+    parser rejects a tree that mixes the two.
+    """
+    candidate = objs.get("flows")
+    if not candidate:
+        return {}
+    if not all(isinstance(obj, Flow) for obj in candidate.values()):
+        return {}
+    return candidate
+
+
+# The SDK's name for each role in a flow.
+_REGISTRATION_ROLES = {"starts_on": "Start", "ends_on": "End", "includes": "Member"}
+
+
+def registration_data(flows: Mapping[str, "Flow"]) -> List[Dict[str, Any]]:
+    """
+    The flow registration data the Glean SDK loads at initialization, in a
+    form the code generators can render directly.
+
+    Flows are emitted as authored, flow to events, in name order. Each event
+    carries its role and whether it was resolved; the SDK builds the reverse
+    index itself.
+    """
+    return [
+        {
+            "name": flow.name,
+            "description": flow.description,
+            "events": [
+                {
+                    "identifier": event,
+                    "role": _REGISTRATION_ROLES[role],
+                    "unresolved": event in flow.unresolved,
+                }
+                for event, role in sorted(flow.events().items())
+            ],
+            "auto_close_timeout_ms": flow.auto_close_timeout_ms,
+            "disabled": flow.disabled,
+        }
+        for _, flow in sorted(flows.items())
+    ]

glean_parser/kotlin.py

@@ -14,6 +14,7 @@
 from typing import Any, Dict, List, Optional, Union  # noqa
 
 from . import __version__
+from . import flows
 from . import metrics
 from . import pings
 from . import util
@@ -231,6 +232,22 @@ def output_kotlin(
             )
             fd.write("\n")
 
+    # Write out the flow registration data, only if there are any flows. The
+    # filename starts with "Glean" so it can never clash with a metric category.
+    flow_data = flows.registration_data(options.get("flows", {}))
+    if flow_data:
+        flows_template = util.get_jinja2_template("kotlin.flows.jinja2")
+        with (output_dir / "GleanFlows.kt").open("w", encoding="utf-8") as fd:
+            fd.write(
+                flows_template.render(
+                    parser_version=__version__,
+                    namespace=namespace,
+                    glean_namespace=glean_namespace,
+                    flows=flow_data,
+                )
+            )
+            fd.write("\n")
+
     template = util.get_jinja2_template(
         "kotlin.jinja2",
         filters=(

glean_parser/lint.py

@@ -20,6 +20,7 @@
 )  # noqa
 
 
+from . import flows
 from . import metrics
 from . import parser
 from . import pings
@@ -494,6 +495,41 @@ def check_name_too_similar(
     "NO_INFO_FOR_OHTTP": (check_ohttp_and_info, CheckType.error),
 }
 
+
+def check_empty_ends_on(
+    flow: flows.Flow, parser_config: Dict[str, Any]
+) -> LintGenerator:
+    if not flow.ends_on:
+        yield (
+            "The flow has no `ends_on` events, so it can only close by timing out "
+            "or when its session ends, and can never record a conclusive outcome. "
+            "A second attempt is also absorbed into the instance that is still open."
+        )
+
+
+def check_unresolved_flow_event(
+    flow: flows.Flow, parser_config: Dict[str, Any]
+) -> LintGenerator:
+    roles = flow.events()
+    for event in sorted(flow.unresolved):
+        yield (
+            f"'{event}' in `{roles[event]}` is not defined in the metrics given to "
+            "glean_parser, so it is assumed to be defined elsewhere, such as in a "
+            "library built separately. Its existence, type, and `in_session` setting "
+            "were not checked. If it is a typo, it will never match; a test that "
+            "records it and checks its `flows` entry will catch that."
+        )
+
+
+# The checks that operate on individual flows:
+#     {NAME: (function, is_error)}
+FLOW_CHECKS: Dict[
+    str, Tuple[Callable[[flows.Flow, dict], LintGenerator], CheckType]
+] = {
+    "EMPTY_ENDS_ON": (check_empty_ends_on, CheckType.warning),
+    "UNRESOLVED_FLOW_EVENT": (check_unresolved_flow_event, CheckType.warning),
+}
+
 ALL_OBJECT_CHECKS: Dict[
     str,
     Tuple[
@@ -546,7 +582,7 @@ def _lint_item_tags(
 
 
 def _lint_pings(
-    category: Dict[str, Union[metrics.Metric, pings.Ping, tags.Tag]],
+    category: Dict[str, Union[metrics.Metric, pings.Ping, tags.Tag, flows.Flow]],
     parser_config: Dict[str, Any],
     valid_tag_names: List[str],
 ) -> List[GlinterNit]:
@@ -578,8 +614,26 @@ def _lint_pings(
     return nits
 
 
+def _lint_flows(
+    category: Dict[str, flows.Flow], parser_config: Dict[str, Any]
+) -> List[GlinterNit]:
+    nits: List[GlinterNit] = []
+
+    for flow_name, flow in sorted(category.items()):
+        for check_name, (check_func, check_type) in FLOW_CHECKS.items():
+            if check_name in flow.no_lint:
+                continue
+            nits.extend(
+                GlinterNit(check_name, flow_name, msg, check_type)
+                for msg in check_func(flow, parser_config)
+            )
+    return nits
+
+
 def _lint_all_objects(
-    objects: Dict[str, Dict[str, Union[metrics.Metric, pings.Ping, tags.Tag]]],
+    objects: Dict[
+        str, Dict[str, Union[metrics.Metric, pings.Ping, tags.Tag, flows.Flow]]
+    ],
     parser_config: Dict[str, Any],
 ) -> List[GlinterNit]:
     nits: List[GlinterNit] = []
@@ -644,6 +698,11 @@ def lint_metrics(
             # currently we have no linting for tags
             continue
 
+        category_flows = flows.flows_in({category_name: category})
+        if category_flows:
+            nits.extend(_lint_flows(category_flows, parser_config))
+            continue
+
         # The information for whether there's duplicate metrics found within the
         # same YAML document is on the category value, which is not presently linted
         # (we lint only its metrics).