feat(highwind): read metrics from metric-hub

jobs/highwind/Dockerfile

@@ -17,6 +17,12 @@ RUN groupadd --gid ${USER_ID} ${GROUP_ID} && \
 
 WORKDIR ${HOME}
 
+# git: metric-config-parser's ConfigCollection clones mozilla/metric-hub at runtime (via GitPython)
+# to read its nimbus/ metric definitions, and python:3.11-slim carries no git binary by default.
+RUN apt-get update && \
+    apt-get install --no-install-recommends --yes git && \
+    rm -rf /var/lib/apt/lists/*
+
 # The literal python3.11 rather than an interpolated version: the base image sets its own
 # PYTHON_VERSION to a full 3.11.x, which shadows a build ARG of the same name and leaves the COPY
 # pointing at a path that does not exist.

jobs/highwind/README.md

@@ -147,8 +147,9 @@ carries no interval, so an inferred schema would vary with the day's mix of stat
 ## Proof-of-concept scope
 
 - Firefox Desktop only.
-- A hard-coded guardrail metric set, so the prototype has no cross-repo dependency. In the real
-  system these come from metric-hub through metric-config-parser.
+- Each experiment's own metric set, read off the Experimenter mirror and resolved against
+  metric-hub's `nimbus/` definitions; a name with no `nimbus/` definition, or whose windows Highwind
+  cannot express, is reported rather than computed.
 - A full recompute every run. Nothing is accumulated across days.
 - Rollouts excluded. They have no randomized control to compare against until the synthetic-control
   work lands, and Jetstream skips them today.

jobs/highwind/highwind/analysis.py

@@ -13,8 +13,9 @@
 The entry point is main.py, which parses the arguments Airflow passes and calls `run_daily_job`
 here.
 
-Desktop experiments only, guardrail metrics only, full recompute every run. Each of those is a
-deliberate limit of the proof of concept and is noted where it bites.
+The job analyses Firefox Desktop experiments only and recomputes every result on each run. Each
+experiment's metrics are the list the Experimenter mirror resolves for it (`discovery.py`), computed
+from their metric-hub `nimbus/` definitions (`metrics.py`).
 """
 
 import contextlib
@@ -25,8 +26,13 @@
 from google.cloud import bigquery
 
 from . import discovery, gbstats_compute
-from . import metrics as metric_definitions_module
 from . import output_writing, sql_generation, sql_running
+from .metrics import (
+    build_aggregation_plan,
+    load_metric_hub,
+    metrics_for_experiment,
+    split_resolved_metrics,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -87,17 +93,17 @@ def run_daily_job(
         if not experiments:
             return []
 
-        metrics_by_source = metric_definitions_module.metric_definitions()
+        metrics_by_slug, resolved_metrics, sources = resolve_run_metrics(experiments)
         windows_by_slug = {
-            experiment.slug: run_windows(experiment, as_of, metrics_by_source)
+            experiment.slug: run_windows(experiment, as_of, metrics_by_slug[experiment.slug])
             for experiment in experiments
         }
         run = sql_generation.Run(
             experiments, windows_by_slug, as_of, COVARIATE_DAYS, sample_percent
         )
 
         cells_by_slug, units_by_slug, timings, failure = gather_sufficient_statistics(
-            client, run, metrics_by_source, as_of, validate_only
+            client, run, resolved_metrics, sources, as_of, validate_only
         )
 
         # No lock: `as_completed` yields in this thread, so the accumulation below is
@@ -109,7 +115,7 @@ def run_daily_job(
                     analyze_experiment,
                     experiment,
                     windows_by_slug[experiment.slug],
-                    metrics_by_source,
+                    metrics_by_slug[experiment.slug],
                     cells_by_slug.get(experiment.slug, {}),
                     failure,
                 )
@@ -162,7 +168,7 @@ def run_daily_job(
                 storage_client,
                 as_of,
                 experiments,
-                all_metrics(metrics_by_source),
+                metrics_by_slug,
                 results_by_slug,
                 cells_by_slug,
                 units_by_slug,
@@ -177,7 +183,7 @@ def publish_analyses(
     storage_client,
     as_of,
     experiments,
-    metrics,
+    metrics_by_slug,
     results_by_slug,
     cells_by_slug,
     units_by_slug,
@@ -195,7 +201,7 @@ def publish_analyses(
                 output_writing.build_analysis,
                 experiment,
                 as_of,
-                metrics,
+                metrics_by_slug[experiment.slug],
                 results_by_slug.get(experiment.slug, []),
                 cells_by_slug.get(experiment.slug, {}),
                 units_by_slug.get(experiment.slug, {}),
@@ -314,8 +320,37 @@ def writes_blobs(validate_only, sample_percent, outputs):
     return sample_percent is None or outputs.local_blob_dir is not None
 
 
-def gather_sufficient_statistics(client, run, metrics_by_source, as_of, validate_only):
-    """Run the shared scan: cohort once, then one query per source and unit, split by slug.
+def resolve_run_metrics(experiments):
+    """Resolve the whole run's metrics from metric-hub, once, from every experiment's own names.
+
+    Returns each experiment's own metric list (for `run_windows`, the statistics and the blob), the
+    resolved metrics by name (for `sql_generation.build_queries`, which scopes them to each
+    analysis unit's own experiments and groups them by source there), and the plan's source
+    definitions (for the SQL itself). A name that does not resolve, whether metric-hub has no
+    `nimbus/` definition for it or its windows are not ones Highwind can express, is reported
+    against every experiment that asked for it rather than failing the run or that experiment.
+    """
+    configs = load_metric_hub()
+    requested = {name for experiment in experiments for name in experiment.metric_names}
+    plan = build_aggregation_plan(configs, requested)
+    resolved_metrics, unresolved_metrics = split_resolved_metrics(plan)
+    metrics_by_slug = {}
+    for experiment in experiments:
+        exp_metrics, problems = metrics_for_experiment(
+            experiment, resolved_metrics, unresolved_metrics
+        )
+        metrics_by_slug[experiment.slug] = exp_metrics
+        for problem in problems:
+            logger.warning(
+                f"{experiment.slug}: metric {problem.name!r} not computed: {problem.reason}",
+                extra={"experiment_slug": experiment.slug, "metric": problem.name},
+            )
+    return metrics_by_slug, resolved_metrics, plan.sources
+
+
+def gather_sufficient_statistics(client, run, resolved_metrics, sources, as_of, validate_only):
+    """Run the shared scan: cohort once, then one query per (source, unit) pair with something to
+    compute, split by slug.
 
     Returns the cells keyed by slug plus whatever went wrong, rather than raising, so a source
     failure becomes an error grid on every experiment instead of an exception that ends the run
@@ -327,7 +362,7 @@ def gather_sufficient_statistics(client, run, metrics_by_source, as_of, validate
         cohort_table, cohort_timing = sql_running.materialize_cohort(
             client, sql_generation.cohort_query(run), as_of, validate_only
         )
-        queries = sql_generation.build_queries(run, metrics_by_source, cohort_table)
+        queries = sql_generation.build_queries(run, resolved_metrics, sources, cohort_table)
         cells_by_slug, timings = sql_running.run_queries(
             client, queries, validate_only=validate_only
         )
@@ -380,23 +415,22 @@ def report_progress(summary, done, total):
     )
 
 
-def analyze_experiment(experiment, windows, metrics_by_source, cells, shared_failure):
+def analyze_experiment(experiment, windows, metrics, cells, shared_failure):
     """One experiment's statistics, from cells the shared scan already produced.
 
     Wrapped so a failure is recorded against this experiment's cells and the loop continues. The
     unit of isolation is the experiment because that is the unit a user reads: one experiment whose
-    statistics blow up should not cost the others their results.
+    statistics blow up should not cost the others their results. `metrics` is this experiment's own
+    resolved list, not the run's full set: two experiments can ask for different metrics.
     """
     try:
         if shared_failure is not None:
-            return failed_summary(experiment, windows, metrics_by_source, shared_failure)
+            return failed_summary(experiment, windows, metrics, shared_failure)
         if not windows:
             # Younger than the shortest window, so no unit has completed one and there is nothing
             # to compute. Its cells are all `not_started` by construction.
             return experiment_summary(experiment, []), []
-        results = gbstats_compute.compute_statistics(
-            experiment, all_metrics(metrics_by_source), windows, cells
-        )
+        results = gbstats_compute.compute_statistics(experiment, metrics, windows, cells)
         return experiment_summary(experiment, results), results
     # The whole point is to record it, not to raise.
     except Exception as error:
@@ -406,7 +440,7 @@ def analyze_experiment(experiment, windows, metrics_by_source, cells, shared_fai
             f"{experiment.slug} failed", extra={"experiment_slug": experiment.slug}
         )
         try:
-            return failed_summary(experiment, windows, metrics_by_source, error)
+            return failed_summary(experiment, windows, metrics, error)
         except Exception as while_recording:
             # A failure while recording a failure must not propagate: it would come back out of
             # future.result() and cost every other experiment its results, which is exactly the
@@ -425,39 +459,36 @@ def analyze_experiment(experiment, windows, metrics_by_source, cells, shared_fai
             )
 
 
-def failed_summary(experiment, windows, metrics_by_source, error):
+def failed_summary(experiment, windows, metrics, error):
     """Record a query-level failure across every cell this experiment expected to produce.
 
     Without this the error rate cannot see an outage: an experiment that produced nothing would
     write no rows and so contribute nothing to the denominator, making total failure look like
     perfect health.
     """
     results = gbstats_compute.compute_statistics(
-        experiment, all_metrics(metrics_by_source), windows, cells={}, failure=error
+        experiment, metrics, windows, cells={}, failure=error
     )
     return experiment_summary(experiment, results, error=error), results
 
 
-def run_windows(experiment, as_of, metrics_by_source=None):
-    """List every distinct window any metric declares, out to the maturity frontier."""
+def run_windows(experiment, as_of, metrics):
+    """List every distinct window any of the experiment's own metrics declares.
+
+    `metrics` is that experiment's own resolved list. Only windows that end within the
+    experiment's tenure as of `as_of` are listed, because no unit can have completed a later
+    window.
+    """
     tenure = experiment.tenure_days(as_of)
-    metrics_by_source = (
-        metrics_by_source or metric_definitions_module.metric_definitions()
-    )
     seen, windows = set(), []
-    for metric in all_metrics(metrics_by_source):
+    for metric in metrics:
         for window in discovery.windows_for(metric, tenure):
             if window.label not in seen:
                 seen.add(window.label)
                 windows.append(window)
     return windows
 
 
-def all_metrics(metrics_by_source):
-    """Flatten the per-source metric mapping into one list."""
-    return [metric for metrics in metrics_by_source.values() for metric in metrics]
-
-
 def experiment_summary(experiment, results, error=None):
     """Summarise what this experiment produced, for the run report and the health measurement.
 

jobs/highwind/highwind/discovery.py

@@ -6,7 +6,7 @@
 """
 
 import datetime
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 
 from google.cloud import bigquery
 
@@ -57,7 +57,18 @@
   start_date,
   end_date,
   reference_branch,
-  ARRAY(SELECT branch.slug FROM UNNEST(branches) AS branch) AS branch_slugs
+  ARRAY(SELECT branch.slug FROM UNNEST(branches) AS branch) AS branch_slugs,
+  -- The experiment's own resolved Jetstream metric set (guardrails, outcomes and custom configs
+  -- alike), already computed for the mirror and `as_of`-matched to Jetstream. NULL for a rollout,
+  -- which `UNNEST` turns into zero rows rather than an error, and which this query excludes anyway.
+  ARRAY(
+    SELECT AS STRUCT
+      metric.name,
+      metric.friendly_name,
+      metric.description,
+      metric.bigger_is_better
+    FROM UNNEST(metric_config.metrics) AS metric
+  ) AS metric_configs
 FROM (
   SELECT
     mirror.* EXCEPT (end_date),
@@ -86,6 +97,19 @@
 """
 
 
+@dataclass(frozen=True)
+class MetricDisplay:
+    """The display text the mirror carried for one of an experiment's metrics.
+
+    Read off the mirror's own resolved Jetstream metric set rather than hard-coded, so the text
+    matches whatever metric-hub says for the experiment that was actually analysed.
+    """
+
+    friendly_name: str | None
+    description: str | None
+    bigger_is_better: bool | None
+
+
 @dataclass(frozen=True)
 class Experiment:
     """One recipe to analyse, with everything the SQL needs and nothing it does not."""
@@ -96,12 +120,21 @@ class Experiment:
     reference_branch: str
     treatment_branches: tuple[str, ...]
     unit: units.AnalysisUnit
+    # The experiment's own metric names, deduplicated, keyed to their display text. Resolving a
+    # name against metric-hub's `nimbus/` definitions is `metrics.py`'s job, not discovery's: this
+    # is only ever what the mirror's own resolved Jetstream metric set names.
+    metrics: dict[str, MetricDisplay] = field(default_factory=dict)
 
     @property
     def branches(self):
         """Return every branch, reference first."""
         return (self.reference_branch, *self.treatment_branches)
 
+    @property
+    def metric_names(self):
+        """Return the experiment's own metric names, sorted for a deterministic order."""
+        return tuple(sorted(self.metrics))
+
     def ended(self, as_of):
         return self.end_date is not None and self.end_date <= as_of
 
@@ -112,12 +145,29 @@ def too_old(self, as_of):
     def tenure_days(self, as_of):
         """Calendar days from first enrollment to the last date with complete data.
 
-        The maturity frontier is derived from this: a window can only be reported once some unit has
-        lived through it, and no unit can have lived longer than the experiment has run.
+        Every window this experiment can report is bounded by this value: a window can only be
+        reported once some unit has lived through it, and no unit can have lived longer than the
+        experiment itself has run.
         """
         return (as_of - self.start_date).days
 
 
+def metric_display(metric_configs):
+    """Turn the mirror's per-experiment metric rows into a name-keyed display map.
+
+    `metric_config.metrics` holds one entry per metric (see `DISCOVERY_SQL`), so this is a
+    straight rekey rather than a dedup.
+    """
+    return {
+        metric.name: MetricDisplay(
+            friendly_name=metric.friendly_name,
+            description=metric.description,
+            bigger_is_better=metric.bigger_is_better,
+        )
+        for metric in metric_configs
+    }
+
+
 def discover(client, as_of, limit=None, only_slugs=None):
     """Experiments to analyse on `as_of`, read from the mirror.
 
@@ -167,6 +217,7 @@ def discover(client, as_of, limit=None, only_slugs=None):
             reference_branch=row.reference_branch,
             treatment_branches=others,
             unit=unit,
+            metrics=metric_display(row.metric_configs),
         )
         if experiment.too_old(as_of):
             skipped.append(
@@ -205,7 +256,8 @@ def length(self):
 
 
 def generate_windows(rule, tenure_days):
-    """Concrete windows for one metric's window rule, up to the maturity frontier.
+    """Concrete windows for one metric's window rule, stopping once a window would need more than
+    `tenure_days` to complete.
 
     Windows are declared as a length and a kind rather than as enumerated bounds, because enumerated
     bounds terminate the series: a metric declaring four weekly windows goes quiet after week four