From 2fa5baf5b81b728f9461f2d018fa0cd48832b16a Mon Sep 17 00:00:00 2001 From: Jared Lockhart <119884+jaredlockhart@users.noreply.github.com> Date: Mon, 26 Oct 2026 18:19:31 -0400 Subject: [PATCH 1/3] feat(highwind): read metrics from metric-hub Because * Highwind's metrics and data sources are hard-coded, so they drift from the definitions Jetstream reads * Each experiment's resolved metric list is already computed in the Experimenter mirror This commit * Reads each experiment's metric names off the mirror's resolved Jetstream metric set and builds their SQL from metric-hub's nimbus/ definitions via metric-config-parser's aggregation_plan() * Resolves the analysis unit's source column per metric-hub data source instead of per app, and reads windows from the plan's cumulative_window/incremental_window/repeat_windows fields * Reports a metric that has no nimbus/ definition, or whose windows Highwind cannot express, against the experiments that asked for it, rather than computing it * Removes the hard-coded guardrail metrics, data sources and reducers, and installs git in the image so metric-config-parser can clone metric-hub at run time https://mozilla-hub.atlassian.net/browse/EXP-7631 --- jobs/highwind/Dockerfile | 6 + jobs/highwind/README.md | 5 +- jobs/highwind/highwind/analysis.py | 87 ++-- jobs/highwind/highwind/discovery.py | 57 ++- jobs/highwind/highwind/metrics.py | 480 ++++++-------------- jobs/highwind/highwind/output_writing.py | 4 +- jobs/highwind/highwind/sql_generation.py | 53 +-- jobs/highwind/highwind/units.py | 41 +- jobs/highwind/requirements.txt | 1 + jobs/highwind/tests/test_analysis.py | 54 ++- jobs/highwind/tests/test_discovery.py | 46 +- jobs/highwind/tests/test_gbstats_compute.py | 4 +- jobs/highwind/tests/test_metrics.py | 273 ++++++++--- jobs/highwind/tests/test_output_writing.py | 6 +- jobs/highwind/tests/test_sql_generation.py | 94 +++- jobs/highwind/tests/test_units.py | 24 +- 16 files changed, 695 insertions(+), 540 deletions(-) diff --git a/jobs/highwind/Dockerfile b/jobs/highwind/Dockerfile index 154d07c01..4a60910b1 100644 --- a/jobs/highwind/Dockerfile +++ b/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. diff --git a/jobs/highwind/README.md b/jobs/highwind/README.md index ee9d7659c..736af8c94 100644 --- a/jobs/highwind/README.md +++ b/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. diff --git a/jobs/highwind/highwind/analysis.py b/jobs/highwind/highwind/analysis.py index 6e3aa6f99..379c0681f 100644 --- a/jobs/highwind/highwind/analysis.py +++ b/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 @@ -24,8 +25,7 @@ from google.cloud import storage # type: ignore from google.cloud import bigquery -from . import discovery, gbstats_compute -from . import metrics as metric_definitions_module +from . import discovery, gbstats_compute, metrics from . import output_writing, sql_generation, sql_running logger = logging.getLogger(__name__) @@ -87,9 +87,9 @@ def run_daily_job( if not experiments: return [] - metrics_by_source = metric_definitions_module.metric_definitions() + metrics_by_slug, metrics_by_source, 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( @@ -97,7 +97,7 @@ def run_daily_job( ) cells_by_slug, units_by_slug, timings, failure = gather_sufficient_statistics( - client, run, metrics_by_source, as_of, validate_only + client, run, metrics_by_source, sources, as_of, validate_only ) # No lock: `as_completed` yields in this thread, so the accumulation below is @@ -109,7 +109,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 +162,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 +177,7 @@ def publish_analyses( storage_client, as_of, experiments, - metrics, + metrics_by_slug, results_by_slug, cells_by_slug, units_by_slug, @@ -195,7 +195,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,7 +314,34 @@ 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): +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 grouped by source (for the shared per-source scan), 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 = metrics.load_configs() + requested = {name for experiment in experiments for name in experiment.metric_names} + plan = metrics.load_plan(configs, requested) + resolved_metrics, unresolved_metrics = metrics.resolve(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, metrics.group_by_source(resolved_metrics), plan.sources + + +def gather_sufficient_statistics(client, run, metrics_by_source, sources, as_of, validate_only): """Run the shared scan: cohort once, then one query per source and unit, split by slug. Returns the cells keyed by slug plus whatever went wrong, rather than raising, so a source @@ -327,7 +354,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, metrics_by_source, sources, cohort_table) cells_by_slug, timings = sql_running.run_queries( client, queries, validate_only=validate_only ) @@ -380,23 +407,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 +432,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,7 +451,7 @@ 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 @@ -433,19 +459,19 @@ def failed_summary(experiment, windows, metrics_by_source, error): 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, out to the maturity frontier. + """ 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) @@ -453,11 +479,6 @@ def run_windows(experiment, as_of, metrics_by_source=None): 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. diff --git a/jobs/highwind/highwind/discovery.py b/jobs/highwind/highwind/discovery.py index 26341588a..2612875cd 100644 --- a/jobs/highwind/highwind/discovery.py +++ b/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,21 @@ 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. + -- Grouped by name because the mirror carries one row per (metric, statistic), so the same metric + -- name can repeat; ANY_VALUE picks the metric's own display text off whichever row survives. + ARRAY( + SELECT AS STRUCT + metric.name, + ANY_VALUE(metric.friendly_name) AS friendly_name, + ANY_VALUE(metric.description) AS description, + ANY_VALUE(metric.bigger_is_better) AS bigger_is_better + FROM UNNEST(metric_config.metrics) AS metric + GROUP BY metric.name + ) AS metric_configs FROM ( SELECT mirror.* EXCEPT (end_date), @@ -86,6 +100,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 +123,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 @@ -118,6 +154,22 @@ def tenure_days(self, as_of): 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. + + The mirror has already deduplicated by name (see `DISCOVERY_SQL`'s `GROUP BY metric.name`), so + this is a straight rekey rather than a second 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 +219,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( diff --git a/jobs/highwind/highwind/metrics.py b/jobs/highwind/highwind/metrics.py index 89b0fbd8b..c7da8cd7c 100644 --- a/jobs/highwind/highwind/metrics.py +++ b/jobs/highwind/highwind/metrics.py @@ -1,382 +1,164 @@ """SECTION 1: METRIC DEFINITIONS. -The Firefox Desktop production default guardrail set, hard-coded. In the real system these come -from metric-hub through metric-config-parser; here they are literal so the proof of concept has no -cross-repo dependency. The shapes below are what the declarative format has to be able to express, -so this file doubles as the target for that work. - -Each metric carries three things: which source table it reads, how to reduce a unit's rows inside a -window to one number, and which windows it declares. +Metrics and the data sources they read come from metric-hub's `nimbus/` definitions, resolved +through metric-config-parser's `aggregation_plan()`. Each experiment names its own metric set (see +`discovery.py`, which reads it off the Experimenter mirror's already-resolved `metric_config`), and +this module turns the union of every experiment's names, for one run, into the reducers and window +rules `sql_generation.py` composes into one shared scan per source. + +A metric-hub metric's aggregation shape (which source it reads, how a unit's rows reduce to one +number) does not depend on which experiment asked for it, so it is resolved once per run. A metric's +windows are declared the same way: metric-hub's `cumulative_window` and `incremental_window` are day +counts, carried over as Highwind's own cumulative and disjoint window rules. Only a *repeating* +family has a Highwind equivalent (see `sql_generation.py`'s window-rule docstring), so a metric +whose `repeat_windows` is false, or that declares neither window, has no window rule Highwind can +build and is reported unresolved rather than silently dropped or mis-rendered. """ from dataclasses import dataclass -from typing import Callable - -# Window rules are a KIND and a LENGTH, never enumerated bounds. Enumerated bounds terminate the -# series: four weekly windows go quiet after week four even though the experiment runs for months. -# A rule extends for as long as the experiment does. Length 7 for both families keeps a young -# experiment producing something in its first week while still extending indefinitely. -CUMULATIVE_WEEKLY = {"kind": "cumulative", "length": 7} -DISJOINT_WEEKLY = {"kind": "disjoint", "length": 7} - -# Retention is a 0/1 "was the unit active in this window", so a cumulative window is ~1.0 by -# construction and says nothing. Those metrics take disjoint weeks instead. -# `is_pinned` and `is_default_browser` are also 0/1 and also monotone over a cumulative window, so -# the same argument arguably applies to them. They are left cumulative to match the desktop -# guardrail set being ported; whether they should move is a question for the metric definitions. -RETENTION_METRICS = {"retained", "retained_dau", "active_in_last_3_days_legacy"} - - -# ------------------------------------------------------------------- per-unit reducers ---- -# A reducer is how one unit's rows become one number, split into composable parts because the -# aggregation reduces a unit's rows once per disjoint bucket and then composes buckets into -# windows: -# -# bucket_aggregate reduces the unit's rows inside ONE bucket to a raw number -# combine how those raw numbers compose over the buckets a window spans, SUM or MIN -# no_rows the raw value a unit with no rows in the window takes -# finalize turns the combined raw number into the metric's value -# -# The split exists to keep the 0/1 metrics correct. A threshold must be applied ONCE, to the -# combined raw value, never per bucket: `retained` carries a SUM per bucket and tests `> 0` at the -# end, because per-bucket thresholds combined afterwards answer a different question as soon as the -# column can be negative or the combination is anything but an OR. - - -def _identity(combined): - return combined - - -def _above_zero(combined): - return f"CAST({combined} > 0 AS INT64)" - - -@dataclass(frozen=True) -class Reducer: - """How one unit's rows in one window become one number, in bucket-composable parts.""" - - bucket_aggregate: str - combine: str = "SUM" - no_rows: str = "0" - finalize: Callable[[str], str] = _identity - - -def per_unit_distinct_days(): - """Distinct days the unit appears inside the window. - - Safe to SUM across buckets only because buckets partition tenure: a submission_date falls in - exactly one bucket, so per-bucket distinct counts add up to the window's distinct count with no - double counting. A combiner that summed overlapping ranges would be wrong here. - """ - return Reducer(bucket_aggregate="COUNT(DISTINCT submission_date)") - - -def per_unit_sum(column): - """Sum of `column` across the unit's rows inside the window.""" - return Reducer(bucket_aggregate=f"SUM({column})") - - -def per_unit_any(condition): - """1 if `condition` held on any of the unit's rows inside the window, else 0. - - Carries the COUNTIF per bucket and tests it once at the end, so the bucket grid never sees the - threshold. This 0/1 reduction is what lets binary metrics travel the same sufficient-statistics - path as the continuous ones, as a linear probability model. - """ - return Reducer(bucket_aggregate=f"COUNTIF({condition})", finalize=_above_zero) - - -def per_unit_countif(condition): - """Count the unit's rows inside the window on which `condition` held. - - Rows, which is what metric-hub specifies and what production computes. The sources are one row - per client per day, so a unit carrying several clients counts one calendar day once per client, - and a row count can exceed the distinct-day count of a metric it nominally contains. The - distinct-day form was measured instead and rejected: each exact COUNT(DISTINCT) is its own - shuffle in the stage that carries essentially all of the cost, for no change in bytes scanned. - """ - return Reducer(bucket_aggregate=f"COUNTIF({condition})") - - -def per_unit_sum_positive(column): - """1 if the unit's summed `column` over the window is positive, else 0. - - The raw SUM is what travels, and `> 0` is applied to the summed total. Testing each bucket and - combining the answers is a different metric: it asks whether any single bucket was positive, - which diverges from the window total the moment `column` can be negative. - - `no_rows` keeps metric-hub's 0-fill. SUM returns NULL when a unit has no rows in the window, - and that NULL would otherwise make the unit's value NULL rather than 0, which the rollup drops - from the numerator while still counting the unit in `n`. - """ - return Reducer(bucket_aggregate=f"SUM({column})", finalize=_above_zero) - -def per_unit_min_below(expression, threshold, no_rows_value): - """1 if the minimum of `expression` over the unit's rows in the window is below `threshold`. +from metric_config_parser.aggregation_plan import aggregation_plan +from metric_config_parser.config import ConfigCollection - Combines with MIN rather than SUM, since the minimum over a window is the minimum of its - buckets' minimums. `no_rows_value` is the sentinel a unit with no rows in the window gets, - applied after the combination for the same reason the thresholds are. - """ - return Reducer( - bucket_aggregate=f"MIN({expression})", - combine="MIN", - no_rows=str(no_rows_value), - finalize=lambda combined: f"CAST({combined} < {threshold} AS INT64)", - ) - - -def per_unit_scaled(reducer, factor): - """Multiply `reducer` by a constant. - - The constant cancels in everything reported here: it divides out of a relative difference, and - out of theta, where it scales numerator and denominator alike. So a scaled metric and its - unscaled twin produce identical intervals, differing only in the raw sums. Both are kept because - both are in the set being ported, and production tells them apart by reading each with a - different statistic; under one uniform method that distinction disappears. - """ - return Reducer( - bucket_aggregate=reducer.bucket_aggregate, - combine=reducer.combine, - no_rows=reducer.no_rows, - finalize=lambda combined: f"({reducer.finalize(combined)}) * {factor}", - ) +APP_NAME = "firefox_desktop" +CUMULATIVE = "cumulative" +DISJOINT = "disjoint" -# ------------------------------------------------------------------------ definitions ---- +# The only lengths metric-hub's desktop guardrails declare today. Kept as named fixtures because +# every window rule built from the plan is one of these two shapes; nothing downstream depends on +# the names themselves. +CUMULATIVE_WEEKLY = {"kind": CUMULATIVE, "length": 7} +DISJOINT_WEEKLY = {"kind": DISJOINT, "length": 7} @dataclass(frozen=True) class Metric: - """One metric: where it reads from, how it reduces a unit's rows, and its windows.""" + """One metric: which metric-hub source it reads, how it reduces a unit's rows, its windows, + and the display text the experiment that asked for it carried in the mirror.""" name: str source: str - reducer: Reducer - window_rules: tuple = (CUMULATIVE_WEEKLY,) + reducer: object + window_rules: tuple friendly_name: str | None = None description: str | None = None -SEARCH_DOCS = "https://docs.telemetry.mozilla.org/datasets/search.html" -DAU_DOCS = ( - "https://mozilla-hub.atlassian.net/wiki/spaces/DATA/pages/314704478/" - "Daily+Active+Users+DAU+Metric" -) +@dataclass(frozen=True) +class UnresolvedMetric: + """A metric name an experiment asked for that this run could not compute, and why.""" -DISPLAY = { - "days_of_use": ( - "Days of use", - "The number of days in the interval that each client sent a main ping.", - ), - "qualified_cumulative_days_of_use": ( - "QCDOU", - "The number of days in the interval that each client sent a main ping, given that the " - "client had >0 active hours and >0 URIs loaded.", - ), - "active_hours": ( - "Active hours", - "Measures the amount of time (in 5-second increments) during which Firefox received user " - "input from a keyboard or mouse. The Firefox window does not need to be focused.", - ), - "uri_count": ( - "URIs visited", - "Counts the total number of URIs visited. Includes within-page navigation events (e.g. to " - "anchors).", - ), - "is_pinned": ( - "Is Pinned (Windows Taskbar)", - "Was Firefox pinned to the Windows Taskbar at any point during the interval?", - ), - "is_default_browser": ( - "Is Default Browser", - "Was Firefox the default browser at any point during the interval?", - ), - "retained": ( - "Retained", - "Records whether a client submitted any pings (i.e. used Firefox). Note: As of June 2026, " - 'this metric is being deprecated in favor of "Retained (DAU)", which better matches the ' - 'conventional definition of Retention based on "active" instead of "seen" now recommended ' - "by Data Science team and used for reporting in other contexts.", - ), - "search_count": ( - "SAP searches", - "Counts the number of searches a user performed through Firefox's Search Access Points. " - f"Learn more in the [search data documentation]({SEARCH_DOCS}).", - ), - "ad_clicks": ( - "Ad clicks", - "Counts clicks on ads on search engine result pages with a Mozilla partner tag.", - ), - "searches_with_ads": ( - "Search result pages with ads", - "Counts search result pages served with advertising. Users may not actually see these ads " - "thanks to e.g. ad-blockers. Learn more in the [search analysis documentation]" - "(https://mozilla-private.report/search-analysis-docs/book/in_content_searches.html).", - ), - "organic_search_count": ( - "Organic searches", - "Counts organic searches, which are searches that are _not_ performed through a Firefox " - "SAP and which are not monetizable. Learn more in the [search data documentation]" - f"({SEARCH_DOCS}).", - ), - "tagged_search_count": ( - "Tagged SAP searches", - "Counts the number of searches a user performed through Firefox's Search Access Points " - "that were submitted with a partner code and were potentially revenue-generating. Learn " - f"more in the [search data documentation]({SEARCH_DOCS}).", - ), - "tagged_follow_on_search_count": ( - "Tagged follow-on searches", - "Counts the number of follow-on searches with a Mozilla partner tag. These are additional " - "searches that users performed from a search engine results page after executing a tagged " - f"search through a SAP. Learn more in the [search data documentation]({SEARCH_DOCS}).", - ), - "retained_dau": ( - "Retained (DAU)", - "Whether the client had at least one DAU-qualifying day in the analysis window (is_dau = " - "TRUE on any day). Conventionally expressed as a percentage rate: The percentage of " - "clients from the originating cohort that were then active in the later period. Most " - "typically, this is measured as Week 2 Retention in order to balance timeliness and " - "accuracy. But when time permits, Data Science recommends using Week 4 Retention as more " - "representative of long-term effects.", - ), - "active_in_last_3_days_legacy": ( - "3 Days Retention", - "Records whether a client submitted any pings (i.e. used Firefox) on any of the last 3 " - "days. Uses legacy telemetry.", - ), - "client_level_daily_active_users_v2": ( - "Firefox Desktop Client-Level DAU", - f"Client-level DAU. The logic is [detailed on the Confluence DAU page]({DAU_DOCS}) and is " - "automatically cross-checked, actively monitored, and change controlled. This metric " - "needs to be aggregated by `submission_date`. If it is not aggregated by " - '`submission_date`, it is similar to a "days of use" metric, and not DAU.', - ), - "daily_active_users_per_1000_clients_legacy": ( - "DAU per 1,000 clients", - f"This metric uses our [canonical, supported definition of Daily Active Users (DAU)]" - f"({DAU_DOCS}), expressed in a format suited for use in experiments. In an experimental " - "comparison, it describes the *additional (incremental) DAU* seen on each day from a " - "treatment. The units are expressed in terms of thousands of clients enrolled or exposed, " - "so the effect can be scaled to either the observed or expected rollout population as " - "needed to estimate absolute DAU impact. Effects are averaged to a per-day basis over " - 'each analysis period. Since feature changes often show a strong "novelty effect", this ' - "metric is best interpreted over Week 4 or later, in order to better estimate what the " - "lasting steady-state effects are.", - ), -} + name: str + reason: str -def _metric(name, source, reducer): - rules = (DISJOINT_WEEKLY,) if name in RETENTION_METRICS else (CUMULATIVE_WEEKLY,) - friendly_name, description = DISPLAY[name] - return Metric( - name=name, - source=source, - reducer=reducer, - window_rules=rules, - friendly_name=friendly_name, - description=description, - ) +def load_configs(): + """Load metric-hub's own `nimbus/` definitions, once per run. + A fresh clone of `mozilla/metric-hub` every call (see metric-config-parser's + `ConfigCollection.from_github_repo`), so the caller does this once for the whole run rather than + once per experiment. + """ + return ConfigCollection.from_github_repo() -CLIENTS_DAILY = "clients_daily" -SEARCH = "search" -ACTIVE_USERS = "active_users" - -GUARDRAILS = [ - _metric("days_of_use", CLIENTS_DAILY, per_unit_distinct_days()), - _metric( - "qualified_cumulative_days_of_use", - CLIENTS_DAILY, - per_unit_countif( - "active_hours_sum > 0 AND " - "scalar_parent_browser_engagement_total_uri_count_normal_and_private_mode_sum > 0" - ), - ), - _metric("active_hours", CLIENTS_DAILY, per_unit_sum("active_hours_sum")), - _metric( - "uri_count", - CLIENTS_DAILY, - per_unit_sum("scalar_parent_browser_engagement_total_uri_count_sum"), - ), - _metric( - "is_pinned", - CLIENTS_DAILY, - per_unit_any("scalar_parent_os_environment_is_taskbar_pinned"), - ), - _metric("is_default_browser", CLIENTS_DAILY, per_unit_any("is_default_browser")), - _metric( - "retained", CLIENTS_DAILY, per_unit_sum_positive("pings_aggregated_by_this_row") - ), - _metric("search_count", SEARCH, per_unit_sum("sap")), - _metric("ad_clicks", SEARCH, per_unit_sum("ad_click")), - _metric("searches_with_ads", SEARCH, per_unit_sum("search_with_ads")), - _metric("organic_search_count", SEARCH, per_unit_sum("organic")), - _metric("tagged_search_count", SEARCH, per_unit_sum("tagged_sap")), - _metric("tagged_follow_on_search_count", SEARCH, per_unit_sum("tagged_follow_on")), - _metric("retained_dau", ACTIVE_USERS, per_unit_any("is_dau")), - _metric( - "active_in_last_3_days_legacy", - ACTIVE_USERS, - per_unit_min_below("mozfun.bits28.days_since_seen(days_active_bits)", 3, 30), - ), - _metric( - "client_level_daily_active_users_v2", ACTIVE_USERS, per_unit_countif("is_dau") - ), - _metric( - "daily_active_users_per_1000_clients_legacy", - ACTIVE_USERS, - per_unit_scaled(per_unit_countif("is_dau"), 1000), - ), -] -# One entry per source table: where to read it, which columns the metrics above need, and any -# restriction the metric-hub data source carries. Keeping the restriction in the source CTE's WHERE -# is the same filter in one scan. The column each table records the analysis unit in comes from the -# unit rather than from here, since it varies by unit and not by table. -SOURCES = { - CLIENTS_DAILY: dict( - table="moz-fx-data-shared-prod.telemetry.clients_daily", - columns=[ - "active_hours_sum", - "scalar_parent_browser_engagement_total_uri_count_sum", - "scalar_parent_browser_engagement_total_uri_count_normal_and_private_mode_sum", - "scalar_parent_os_environment_is_taskbar_pinned", - "is_default_browser", - "pings_aggregated_by_this_row", - ], - ), - SEARCH: dict( - table="moz-fx-data-shared-prod.search.search_clients_engines_sources_daily", - columns=[ - "sap", - "ad_click", - "search_with_ads", - "organic", - "tagged_sap", - "tagged_follow_on", - ], - ), - # metric-hub's firefox_desktop_active_users_view is - # (SELECT * FROM ...telemetry.desktop_active_users WHERE is_desktop). - ACTIVE_USERS: dict( - table="moz-fx-data-shared-prod.telemetry.desktop_active_users", - where="is_desktop", - columns=["is_dau", "days_active_bits"], - ), -} +def load_plan(configs, metric_names): + """Resolve one run's metric names against metric-hub's `nimbus/` definitions for this app.""" + return aggregation_plan(configs, APP_NAME, sorted(set(metric_names))) -def metric_definitions(): - """Group the run's metric set by the source table each one reads. +def window_rules(aggregation_metric): + """The window rules Highwind can build from a plan metric's window fields, or why not. - Grouped because one query per source is what makes adding a metric to an - already-scanned table nearly free, which is the property the cost model rests on. + Returns `(rules, None)` when the metric has at least one window family and repeats it, or + `(None, reason)` otherwise. A metric with neither `cumulative_window` nor `incremental_window` + set has nothing for Highwind to compute a window over at all; one that sets `repeat_windows` + false declares a single window that stops, which none of Highwind's window rules can express, + since every rule it has ever supported extends for as long as the experiment runs (see + `sql_generation.py`'s `Run` docstring). Mapping either case onto a repeating rule would compute + windows the metric never declared. + """ + no_window = ( + aggregation_metric.cumulative_window is None + and aggregation_metric.incremental_window is None + ) + if no_window: + return None, "no window declared" + if not aggregation_metric.repeat_windows: + return None, ( + "repeat_windows is false; Highwind's window rules always repeat and cannot " + "express a one-off window" + ) + rules = [] + if aggregation_metric.cumulative_window is not None: + rules.append({"kind": CUMULATIVE, "length": aggregation_metric.cumulative_window}) + if aggregation_metric.incremental_window is not None: + rules.append({"kind": DISJOINT, "length": aggregation_metric.incremental_window}) + return tuple(rules), None + + +def resolve(plan): + """Split one run's plan into the metrics Highwind can compute and the ones it cannot, with why. + + A name fails to resolve two ways: no `nimbus/` definition at all (`plan.missing`), or a + definition whose windows Highwind has no rule for (`window_rules` above). Both are reported by + name and reason rather than raised, so neither one metric nor an unrelated experiment is lost + over it. + """ + resolved, unresolved = {}, {} + for name in plan.missing: + unresolved[name] = "no nimbus/ definition for this app" + for name, aggregation_metric in plan.metrics.items(): + rules, reason = window_rules(aggregation_metric) + if rules is None: + unresolved[name] = reason + continue + resolved[name] = Metric( + name=name, + source=aggregation_metric.data_source, + reducer=aggregation_metric.aggregation, + window_rules=rules, + ) + return resolved, unresolved + + +def group_by_source(resolved_metrics): + """Group resolved metrics by the metric-hub source they read. + + Grouped because one query per source is what makes adding a metric to an already-scanned table + nearly free, which is the property the cost model rests on (see `sql_generation.py`). """ by_source = {} - for metric in GUARDRAILS: + for metric in resolved_metrics.values(): by_source.setdefault(metric.source, []).append(metric) return by_source + + +def for_experiment(experiment, resolved_metrics, unresolved_metrics): + """One experiment's own metric list, with its own display text, and the names it could not get. + + A metric's aggregation shape is the same for every experiment that reads it, so + `resolved_metrics` is resolved once for the whole run; this keeps only the names the experiment + itself asked for and attaches the friendly name and description the mirror carried for THAT + experiment, which can differ from another experiment's text for a metric of the same name. + """ + chosen, problems = [], [] + for name in experiment.metric_names: + metric = resolved_metrics.get(name) + if metric is None: + reason = unresolved_metrics.get(name, "no nimbus/ definition for this app") + problems.append(UnresolvedMetric(name=name, reason=reason)) + continue + display = experiment.metrics.get(name) + chosen.append( + Metric( + name=metric.name, + source=metric.source, + reducer=metric.reducer, + window_rules=metric.window_rules, + friendly_name=display.friendly_name if display else None, + description=display.description if display else None, + ) + ) + return chosen, problems diff --git a/jobs/highwind/highwind/output_writing.py b/jobs/highwind/highwind/output_writing.py index 5535e7811..d322b9a16 100644 --- a/jobs/highwind/highwind/output_writing.py +++ b/jobs/highwind/highwind/output_writing.py @@ -84,7 +84,7 @@ bigquery.SchemaField( "metric", "STRING", - description="Metric name, from the hard-coded desktop guardrail set.", + description="Metric name, resolved against metric-hub's nimbus/ definitions.", ), bigquery.SchemaField( "window", @@ -212,7 +212,7 @@ bigquery.SchemaField( "metric", "STRING", - description="Metric name, from the hard-coded desktop guardrail set.", + description="Metric name, resolved against metric-hub's nimbus/ definitions.", ), bigquery.SchemaField( "window", diff --git a/jobs/highwind/highwind/sql_generation.py b/jobs/highwind/highwind/sql_generation.py index 7beace00f..c78b7ea51 100644 --- a/jobs/highwind/highwind/sql_generation.py +++ b/jobs/highwind/highwind/sql_generation.py @@ -44,7 +44,6 @@ import math import textwrap -from .metrics import SOURCES from .units import PROFILE_GROUP_ID # The bucket a unit's pre-enrollment rows land in. The covariate is one fixed window rather than @@ -453,24 +452,24 @@ def enrollment_unit_id(run): # ----------------------------------------------------------------------- the source queries ---- -def build_queries(run, metrics_by_source, cohort_table): +def build_queries(run, metrics_by_source, sources, cohort_table): """One SQL string per (source table, analysis unit), covering every experiment in the run. Keyed by both, because a run whose experiments randomize on different units reads each source - once per unit and the two are separate queries with separate costs. + once per unit and the two are separate queries with separate costs. `sources` is the + aggregation plan's resolved source definitions, keyed the same way as `metrics_by_source`. """ return { f"{source}/{unit.kind}": build_source_query( - unit_run, source, metrics, cohort_table, unit + unit_run, source, metrics, sources[source], cohort_table, unit ) for unit, unit_run in run.by_unit().items() for source, metrics in metrics_by_source.items() } -def build_source_query(run, source, metrics, cohort_table, unit): +def build_source_query(run, source, metrics, source_definition, cohort_table, unit): """Build the full query for one source: cohort, per-slug facts, rows, stages, rollup.""" - source_definition = SOURCES[source] return textwrap.dedent(f""" WITH cohort AS ( {_indent(cohort_source(cohort_table, unit))} @@ -527,30 +526,30 @@ def source_cte(source_definition, run, unit): why an old recipe is expensive no matter how the window horizon is set, and why age is handled by declining to analyse it. - Column selection is explicit, so the scan is the columns this source's metrics name plus - whatever a source-level restriction reads. `desktop_active_users` is the case where those - differ: its `is_desktop` is a computed column, so filtering on it also reads the ISP, - distribution and version columns behind it. + `source_definition.from_expression` is used verbatim rather than backtick-quoted, because + metric-hub's own data source can be a parenthesised subquery rather than a bare table name: the + active-users source's `is_desktop` restriction is baked into its `from_expression` by metric-hub + itself, so Highwind applies no restriction of its own here. + + Column selection is explicit, so the scan is the columns this source's metrics name plus the + submission date column every source declares, deduplicated against each other. """ - columns = ",\n ".join(source_definition["columns"]) - restriction = ( - f"\n AND {source_definition['where']}" - if source_definition.get("where") - else "" - ) + submission_date_column = source_definition.submission_date_column + extra_columns = sorted(set(source_definition.columns) - {submission_date_column}) + separator = ",\n " + extra = f"{separator}{separator.join(extra_columns)}" if extra_columns else "" return textwrap.dedent(f""" SELECT - {unit.source_column} AS unit_id, - submission_date, - {columns} - FROM `{source_definition['table']}` - WHERE submission_date + {unit.source_column(source_definition)} AS unit_id, + {submission_date_column} AS submission_date{extra} + FROM {source_definition.from_expression} + WHERE {submission_date_column} BETWEEN DATE '{run.first_source_date()}' - AND DATE '{run.as_of}'{restriction}{source_sample(run, unit)} + AND DATE '{run.as_of}'{source_sample(run, unit, source_definition)} """).strip() -def source_sample(run, unit): +def source_sample(run, unit, source_definition): """Restrict a source scan to the run's sample, in whichever form this unit allows. The bare clustered column where the unit is the id it is hashed from, since only that form lets @@ -561,7 +560,9 @@ def source_sample(run, unit): return "" if unit.clustered_sample_id: return f"\n AND sample_id < {run.sample_percent}" - return sample_clause(unit.source_column, run.sample_percent, prefix="\n AND ") + return sample_clause( + unit.source_column(source_definition), run.sample_percent, prefix="\n AND " + ) # ---------------------------------------------------------------------------- the stages ---- @@ -708,7 +709,7 @@ def window_value(metric, kind): ) else: combined = f"bucket_total.{metric.name}__raw" - return reducer.finalize(f"COALESCE({combined}, {reducer.no_rows})") + return reducer.finalize_sql(f"COALESCE({combined}, {reducer.no_rows})") def covariate_value(metric): @@ -718,7 +719,7 @@ def covariate_value(metric): f"MAX(IF(grid.bucket = {PRE_BUCKET}, bucket_total.{metric.name}__raw, NULL))" f" OVER unit_all" ) - return reducer.finalize(f"COALESCE({combined}, {reducer.no_rows})") + return reducer.finalize_sql(f"COALESCE({combined}, {reducer.no_rows})") def rollup_select(metrics): diff --git a/jobs/highwind/highwind/units.py b/jobs/highwind/highwind/units.py index 214a44917..a084adf9b 100644 --- a/jobs/highwind/highwind/units.py +++ b/jobs/highwind/highwind/units.py @@ -22,23 +22,24 @@ @dataclass(frozen=True) class AnalysisUnit: - """One analysis unit, and the column each side of the join records it in. + """One analysis unit, and the enrollment-side column it is read from. - `enrollment_column` and `source_column` are separate because the two sides of the join record - the same physical unit under different column names, and one of those names is reused for a - different id. On desktop, enrollment events are read from Glean's `events_stream`, where + `enrollment_column` is separate from a source's own id column because the two sides of the join + record the same physical unit under different column names, and one of those names is reused for + a different id. On desktop, enrollment events are read from Glean's `events_stream`, where `client_id` is the Glean client id and the legacy telemetry id sits beside it in - `legacy_telemetry_client_id`. The metric tables are legacy tables, whose own `client_id` column - holds that same legacy telemetry id. So both columns named here mean the legacy id: it is called - `legacy_telemetry_client_id` on the enrollment side and `client_id` on the source side. + `legacy_telemetry_client_id`. The metric-hub sources are legacy tables, whose own `client_id` + column holds that same legacy telemetry id. So both ids mean the legacy one: it is called + `legacy_telemetry_client_id` on the enrollment side and `client_id` on a source. - Joining `events_stream.client_id` to a metric table's `client_id` would therefore compare a - Glean id against a legacy one. That matches nothing while still producing a full cohort, so - every metric would silently sum to zero rather than fail, which is the failure the two columns - exist to make impossible to write by accident. + Joining `events_stream.client_id` to a source's `client_id` would therefore compare a Glean id + against a legacy one. That matches nothing while still producing a full cohort, so every metric + would silently sum to zero rather than fail, which is the failure the two columns exist to make + impossible to write by accident. - One `source_column` serves every source table because an app's tables agree on the name; a - future app whose tables do not can carry the difference on the source instead. + A source's own id column is resolved per source through `source_column` rather than carried + here, because metric-hub names it per data source (`client_id_column`/`group_id_column` on the + resolved `AggregationSource`) rather than once per app. `clustered_sample_id` is whether the source tables' clustered `sample_id` column is hashed from THIS unit. It decides how a sampled run may be restricted, so it is per unit rather than assumed @@ -47,9 +48,18 @@ class AnalysisUnit: kind: str enrollment_column: str - source_column: str clustered_sample_id: bool + def source_column(self, source): + """The column `source` (a metric-hub `AggregationSource`) records this unit's id in. + + A profile-group unit reads the source's group id column, a client unit its client id + column; both are carried on every `AggregationSource` the aggregation plan resolves. + """ + if self.kind == PROFILE_GROUP_ID: + return source.group_id_column + return source.client_id_column + # Keyed on the mirror's `app_name` and the recipe's `bucketConfig.randomizationUnit`, which is the # pair the mirror reports for every recipe. Desktop randomizes on `normandy_id` by default and on @@ -59,19 +69,16 @@ class AnalysisUnit: ("firefox_desktop", "group_id"): AnalysisUnit( kind=PROFILE_GROUP_ID, enrollment_column=PROFILE_GROUP_ID, - source_column=PROFILE_GROUP_ID, clustered_sample_id=False, ), ("firefox_desktop", "normandy_id"): AnalysisUnit( kind=CLIENT_ID, enrollment_column="legacy_telemetry_client_id", - source_column=CLIENT_ID, clustered_sample_id=True, ), ("firefox_desktop", "client_id"): AnalysisUnit( kind=CLIENT_ID, enrollment_column="legacy_telemetry_client_id", - source_column=CLIENT_ID, clustered_sample_id=True, ), } diff --git a/jobs/highwind/requirements.txt b/jobs/highwind/requirements.txt index 918102db9..d3a0a4e79 100644 --- a/jobs/highwind/requirements.txt +++ b/jobs/highwind/requirements.txt @@ -1,6 +1,7 @@ click==8.3.1 google-cloud-bigquery==3.40.0 google-cloud-storage==3.8.0 +mozilla-metric-config-parser==2026.10.2 mozilla-nimbus-schemas==3006.0.0 numpy==1.24.2 pandas==2.3.3 diff --git a/jobs/highwind/tests/test_analysis.py b/jobs/highwind/tests/test_analysis.py index 9ea5d4af6..e0244f75d 100644 --- a/jobs/highwind/tests/test_analysis.py +++ b/jobs/highwind/tests/test_analysis.py @@ -5,6 +5,8 @@ import types import pytest +from metric_config_parser.aggregation import BucketedAggregation, Combine +from metric_config_parser.aggregation_plan import AggregationSource from mozilla_nimbus_schemas.highwind import HighwindAnalysis from highwind import analysis, output_writing, sql_generation, units @@ -16,7 +18,7 @@ INSUFFICIENT_DATA, NOT_STARTED, ) -from highwind.metrics import metric_definitions +from highwind.metrics import CUMULATIVE_WEEKLY, Metric AS_OF = datetime.date(2026, 8, 20) @@ -29,6 +31,34 @@ unit=units.resolve("firefox_desktop", "normandy_id"), ) +# One resolved metric, standing in for what `metrics.resolve()` would build off a real aggregation +# plan: these tests exercise the run's orchestration, not metric resolution itself (see +# test_metrics.py for that). +TEST_METRICS = [ + Metric( + name="active_hours", + source="clients_daily", + reducer=BucketedAggregation( + bucket_aggregate="SUM(active_hours_sum)", + combine=Combine.SUM, + no_rows="0", + finalize="{combined}", + ), + window_rules=(CUMULATIVE_WEEKLY,), + ) +] +TEST_SOURCES = { + "clients_daily": AggregationSource( + data_source="clients_daily", + from_expression="mozdata.telemetry.clients_daily", + client_id_column="client_id", + group_id_column="profile_group_id", + analysis_units=("client_id", "profile_group_id"), + submission_date_column="submission_date", + columns=("active_hours_sum",), + ) +} + class FakeLoadJob: def result(self): @@ -145,7 +175,7 @@ def test_a_failure_while_recording_a_failure_is_still_returned_not_raised(): # has to come back as a summary, because raising here escapes through future.result() and costs # every other experiment its results. recorded, results = analysis.analyze_experiment( - EXPERIMENT, [], {"clients_daily": None}, {}, "the shared scan failed" + EXPERIMENT, [], None, {}, "the shared scan failed" ) assert results == [] @@ -226,7 +256,7 @@ def test_an_experiment_that_fails_outright_is_recorded_against_its_slug(): with analysis.collecting_run_log(AS_OF, client, output_writing.Outputs(), True): analysis.analyze_experiment( - EXPERIMENT, [], {"clients_daily": None}, {}, "the shared scan failed" + EXPERIMENT, [], None, {}, "the shared scan failed" ) failures = [row for row in client.log_rows if row["log_level"] == "ERROR"] @@ -337,7 +367,7 @@ def publish(storage, experiments, refused, results_by_slug, run_log): storage, AS_OF, experiments, - analysis.all_metrics(metric_definitions()), + {experiment.slug: TEST_METRICS for experiment in experiments}, results_by_slug, {}, {}, @@ -357,8 +387,7 @@ def test_analysed_and_refused_experiments_each_get_a_blob_carrying_their_own_log unit=units.resolve("firefox_desktop", "normandy_id"), ) storage, client = RecordingStorage(), RecordingClient() - metrics_by_source = metric_definitions() - windows = analysis.run_windows(EXPERIMENT, AS_OF, metrics_by_source) + windows = analysis.run_windows(EXPERIMENT, AS_OF, TEST_METRICS) with analysis.collecting_run_log( AS_OF, client, output_writing.Outputs(), False @@ -366,16 +395,14 @@ def test_analysed_and_refused_experiments_each_get_a_blob_carrying_their_own_log analysis.report_selection( [EXPERIMENT], [("an-old-slug", "416d old", refused)], AS_OF ) - _, results = analysis.analyze_experiment( - EXPERIMENT, windows, metrics_by_source, {}, None - ) + _, results = analysis.analyze_experiment(EXPERIMENT, windows, TEST_METRICS, {}, None) publish(storage, [EXPERIMENT], [refused], {"a-slug": results}, run_log) assert sorted(storage.uploaded) == ["highwind/a_slug.json", "highwind/an_old_slug.json"] analysed = storage.payloads["highwind/a_slug.json"] refusal = storage.payloads["highwind/an_old_slug.json"] assert [metric.slug for metric in analysed.metrics] == [ - metric.name for metric in analysis.all_metrics(metrics_by_source) + metric.name for metric in TEST_METRICS ] assert analysed.errors == [] assert refusal.metrics == [] @@ -408,7 +435,6 @@ def test_a_blob_that_cannot_be_built_is_logged_rather_than_raised(): unit=units.AnalysisUnit( kind="nimbus_id", enrollment_column="nimbus_id", - source_column="nimbus_id", clustered_sample_id=False, ), ) @@ -474,17 +500,17 @@ def update_table(self, table, fields): def test_a_failed_unit_count_keeps_the_cells_the_shared_scan_computed(caplog): # The unit counts are display-only, so losing them must not turn every experiment's computed # statistics into an error grid. - metrics_by_source = metric_definitions() + metrics_by_source = {"clients_daily": TEST_METRICS} run = sql_generation.Run( [EXPERIMENT], - {"a-slug": analysis.run_windows(EXPERIMENT, AS_OF, metrics_by_source)}, + {"a-slug": analysis.run_windows(EXPERIMENT, AS_OF, TEST_METRICS)}, AS_OF, analysis.COVARIATE_DAYS, ) cells_by_slug, units_by_slug, timings, failure = ( analysis.gather_sufficient_statistics( - UnitCountFailingClient(), run, metrics_by_source, AS_OF, False + UnitCountFailingClient(), run, metrics_by_source, TEST_SOURCES, AS_OF, False ) ) diff --git a/jobs/highwind/tests/test_discovery.py b/jobs/highwind/tests/test_discovery.py index 47b5ca1a1..868f5dd83 100644 --- a/jobs/highwind/tests/test_discovery.py +++ b/jobs/highwind/tests/test_discovery.py @@ -8,7 +8,7 @@ import pytest from highwind import discovery, units -from highwind.metrics import CUMULATIVE_WEEKLY, DISJOINT_WEEKLY, Metric, per_unit_sum +from highwind.metrics import CUMULATIVE_WEEKLY, DISJOINT_WEEKLY, Metric AS_OF = datetime.date(2026, 8, 1) @@ -28,7 +28,7 @@ def metric(*window_rules): return Metric( name="a-metric", source="clients_daily", - reducer=per_unit_sum("a_column"), + reducer=None, window_rules=window_rules, ) @@ -103,6 +103,15 @@ def test_branches_lists_the_reference_first(): assert experiment(AS_OF).branches == ("control", "treatment-a", "treatment-b") +def metric_config(name, friendly_name=None, description=None, bigger_is_better=None): + return types.SimpleNamespace( + name=name, + friendly_name=friendly_name, + description=description, + bigger_is_better=bigger_is_better, + ) + + def mirror_row(**overrides): """One row of the mirror, in the shape `discover` reads.""" return types.SimpleNamespace( @@ -114,6 +123,7 @@ def mirror_row(**overrides): "end_date": None, "reference_branch": "control", "branch_slugs": ["control", "treatment"], + "metric_configs": [], **overrides, } ) @@ -147,6 +157,38 @@ def test_an_experiment_is_analysed_at_the_unit_its_own_recipe_randomized_on(): } +def test_an_experiment_carries_its_own_deduplicated_metric_names(): + experiments, _ = discovery.discover( + FakeClient( + [ + mirror_row( + metric_configs=[ + metric_config("active_hours", friendly_name="Active hours"), + # The mirror carries one row per (metric, statistic), so the same metric + # name can repeat; discover() reads the dedup DISCOVERY_SQL already did. + metric_config("active_hours", friendly_name="Active hours"), + metric_config("retained_dau", friendly_name="Retained (DAU)"), + ] + ) + ] + ), + AS_OF, + ) + + (experiment,) = experiments + assert experiment.metric_names == ("active_hours", "retained_dau") + assert experiment.metrics["active_hours"].friendly_name == "Active hours" + + +def test_a_rollout_with_no_metric_config_carries_no_metrics(): + experiments, _ = discovery.discover( + FakeClient([mirror_row(metric_configs=[])]), AS_OF + ) + + (experiment,) = experiments + assert experiment.metric_names == () + + def test_a_randomization_unit_with_no_analysis_unit_is_skipped_not_defaulted(): # The refusal is the point: analysing at a grain the recipe did not randomize on produces # intervals that look no different from correct ones. diff --git a/jobs/highwind/tests/test_gbstats_compute.py b/jobs/highwind/tests/test_gbstats_compute.py index dc4558e46..8047cfc43 100644 --- a/jobs/highwind/tests/test_gbstats_compute.py +++ b/jobs/highwind/tests/test_gbstats_compute.py @@ -17,13 +17,13 @@ CONTINUOUS = metrics.Metric( name="active_hours", source="clients_daily", - reducer=metrics.per_unit_sum("active_hours_sum"), + reducer=None, window_rules=(metrics.CUMULATIVE_WEEKLY,), ) BINARY = metrics.Metric( name="retained_dau", source="active_users", - reducer=metrics.per_unit_any("is_dau"), + reducer=None, window_rules=(metrics.DISJOINT_WEEKLY,), ) diff --git a/jobs/highwind/tests/test_metrics.py b/jobs/highwind/tests/test_metrics.py index a4876105f..dc982ea18 100644 --- a/jobs/highwind/tests/test_metrics.py +++ b/jobs/highwind/tests/test_metrics.py @@ -1,91 +1,218 @@ -"""The per-unit reducers, whose parts have to compose over buckets without moving a threshold.""" +"""Resolving a run's metric names against metric-hub's aggregation plan.""" -from highwind import metrics - - -def test_a_sum_reduces_with_a_sum_and_finalizes_to_itself(): - reducer = metrics.per_unit_sum("active_hours_sum") - - assert reducer.bucket_aggregate == "SUM(active_hours_sum)" - assert reducer.combine == "SUM" - assert reducer.no_rows == "0" - assert reducer.finalize("combined") == "combined" - - -def test_distinct_days_sums_over_buckets_because_buckets_partition_tenure(): - reducer = metrics.per_unit_distinct_days() - - assert reducer.bucket_aggregate == "COUNT(DISTINCT submission_date)" - assert reducer.combine == "SUM" - assert reducer.finalize("combined") == "combined" - - -def test_any_counts_per_bucket_and_thresholds_the_combined_count(): - reducer = metrics.per_unit_any("is_default_browser") - - assert reducer.bucket_aggregate == "COUNTIF(is_default_browser)" - assert reducer.combine == "SUM" - assert reducer.finalize("combined") == "CAST(combined > 0 AS INT64)" - - -def test_countif_keeps_the_count_rather_than_reducing_it_to_a_flag(): - reducer = metrics.per_unit_countif("is_dau") - - assert reducer.bucket_aggregate == "COUNTIF(is_dau)" - assert reducer.finalize("combined") == "combined" - - -def test_sum_positive_thresholds_the_summed_total_not_each_bucket(): - reducer = metrics.per_unit_sum_positive("pings_aggregated_by_this_row") - - assert reducer.bucket_aggregate == "SUM(pings_aggregated_by_this_row)" - assert reducer.combine == "SUM" - assert reducer.finalize("combined") == "CAST(combined > 0 AS INT64)" +from metric_config_parser.aggregation import BucketedAggregation, Combine +from metric_config_parser.aggregation_plan import AggregationMetric, AggregationPlan +from highwind import metrics +from highwind.discovery import Experiment, MetricDisplay +from highwind import units -def test_min_below_combines_with_min_and_carries_its_own_missing_value(): - reducer = metrics.per_unit_min_below("days_since_seen", 3, 30) +REDUCER = BucketedAggregation( + bucket_aggregate="COALESCE(SUM(active_hours_sum), 0)", + combine=Combine.SUM, + no_rows="0", + finalize="{combined}", +) - assert reducer.bucket_aggregate == "MIN(days_since_seen)" - assert reducer.combine == "MIN" - assert reducer.no_rows == "30" - assert reducer.finalize("combined") == "CAST(combined < 3 AS INT64)" +def aggregation_metric( + data_source="clients_daily", + cumulative_window=7, + incremental_window=None, + repeat_windows=True, +): + return AggregationMetric( + name="active_hours", + data_source=data_source, + aggregation=REDUCER, + cumulative_window=cumulative_window, + incremental_window=incremental_window, + repeat_windows=repeat_windows, + ) -def test_scaling_multiplies_the_finalized_value_and_leaves_the_bucket_aggregate_alone(): - inner = metrics.per_unit_countif("is_dau") - scaled = metrics.per_unit_scaled(inner, 1000) - assert scaled.bucket_aggregate == inner.bucket_aggregate - assert scaled.combine == inner.combine - assert scaled.no_rows == inner.no_rows - assert scaled.finalize("combined") == "(combined) * 1000" +def plan_of(metrics_by_name, missing=()): + return AggregationPlan(sources={}, metrics=metrics_by_name, missing=list(missing)) -def test_scaling_a_binary_metric_thresholds_before_it_multiplies(): - scaled = metrics.per_unit_scaled(metrics.per_unit_any("is_dau"), 1000) +def test_a_cumulative_window_becomes_a_cumulative_rule(): + rules, reason = metrics.window_rules(aggregation_metric(cumulative_window=7)) - assert scaled.finalize("combined") == "(CAST(combined > 0 AS INT64)) * 1000" + assert rules == ({"kind": "cumulative", "length": 7},) + assert reason is None -def test_binary_retention_metrics_take_disjoint_weeks(): - by_name = {metric.name: metric for metric in metrics.GUARDRAILS} +def test_an_incremental_window_becomes_a_disjoint_rule(): + rules, reason = metrics.window_rules( + aggregation_metric(cumulative_window=None, incremental_window=7) + ) - for name in metrics.RETENTION_METRICS: - assert by_name[name].window_rules == (metrics.DISJOINT_WEEKLY,) - assert by_name["active_hours"].window_rules == (metrics.CUMULATIVE_WEEKLY,) + assert rules == ({"kind": "disjoint", "length": 7},) + assert reason is None -def test_every_metric_reads_a_source_the_run_knows_how_to_scan(): - by_source = metrics.metric_definitions() +def test_both_windows_together_become_both_rules(): + rules, reason = metrics.window_rules( + aggregation_metric(cumulative_window=7, incremental_window=3) + ) - assert set(by_source) == set(metrics.SOURCES) - assert sum(len(group) for group in by_source.values()) == len(metrics.GUARDRAILS) - for source, group in by_source.items(): - assert {metric.source for metric in group} == {source} + assert rules == ( + {"kind": "cumulative", "length": 7}, + {"kind": "disjoint", "length": 3}, + ) + assert reason is None -def test_each_source_declares_a_table_and_the_columns_its_metrics_name(): - for spec in metrics.SOURCES.values(): - assert spec["table"].count(".") == 2 - assert spec["columns"] +def test_no_window_at_all_is_unsupported(): + rules, reason = metrics.window_rules( + aggregation_metric(cumulative_window=None, incremental_window=None) + ) + + assert rules is None + assert "no window" in reason + + +def test_a_non_repeating_window_is_unsupported(): + # Highwind's window rules always repeat for the life of the experiment; a metric that declares + # a single window that stops has no equivalent rule. + rules, reason = metrics.window_rules(aggregation_metric(repeat_windows=False)) + + assert rules is None + assert "repeat_windows" in reason + + +def test_resolve_splits_missing_names_into_unresolved_with_a_reason(): + plan = plan_of({}, missing=["no_such_metric"]) + + resolved, unresolved = metrics.resolve(plan) + + assert resolved == {} + assert "no nimbus/ definition" in unresolved["no_such_metric"] + + +def test_resolve_reports_an_unsupported_window_the_same_way_as_a_missing_name(): + plan = plan_of({"active_hours": aggregation_metric(repeat_windows=False)}) + + resolved, unresolved = metrics.resolve(plan) + + assert resolved == {} + assert "repeat_windows" in unresolved["active_hours"] + + +def test_resolve_builds_a_metric_from_a_resolvable_plan_entry(): + plan = plan_of({"active_hours": aggregation_metric()}) + + resolved, unresolved = metrics.resolve(plan) + + assert unresolved == {} + metric = resolved["active_hours"] + assert metric.name == "active_hours" + assert metric.source == "clients_daily" + assert metric.reducer is REDUCER + assert metric.window_rules == ({"kind": "cumulative", "length": 7},) + assert metric.friendly_name is None + assert metric.description is None + + +def test_group_by_source_groups_resolved_metrics_by_their_data_source(): + resolved = { + "active_hours": metrics.Metric( + name="active_hours", + source="clients_daily", + reducer=REDUCER, + window_rules=(metrics.CUMULATIVE_WEEKLY,), + ), + "retained_dau": metrics.Metric( + name="retained_dau", + source="active_users", + reducer=REDUCER, + window_rules=(metrics.DISJOINT_WEEKLY,), + ), + "uri_count": metrics.Metric( + name="uri_count", + source="clients_daily", + reducer=REDUCER, + window_rules=(metrics.CUMULATIVE_WEEKLY,), + ), + } + + by_source = metrics.group_by_source(resolved) + + assert {metric.name for metric in by_source["clients_daily"]} == { + "active_hours", + "uri_count", + } + assert {metric.name for metric in by_source["active_users"]} == {"retained_dau"} + + +def experiment_with(metric_names, display=None): + return Experiment( + slug="a-slug", + start_date=None, + end_date=None, + reference_branch="control", + treatment_branches=("treatment",), + unit=units.resolve("firefox_desktop", "normandy_id"), + metrics={ + name: (display or {}).get( + name, MetricDisplay(friendly_name=None, description=None, bigger_is_better=None) + ) + for name in metric_names + }, + ) + + +def test_for_experiment_keeps_only_the_names_it_asked_for(): + resolved = { + "active_hours": metrics.Metric( + name="active_hours", + source="clients_daily", + reducer=REDUCER, + window_rules=(metrics.CUMULATIVE_WEEKLY,), + ), + "retained_dau": metrics.Metric( + name="retained_dau", + source="active_users", + reducer=REDUCER, + window_rules=(metrics.DISJOINT_WEEKLY,), + ), + } + experiment = experiment_with(["active_hours"]) + + chosen, problems = metrics.for_experiment(experiment, resolved, {}) + + assert [metric.name for metric in chosen] == ["active_hours"] + assert problems == [] + + +def test_for_experiment_attaches_its_own_display_text(): + resolved = { + "active_hours": metrics.Metric( + name="active_hours", + source="clients_daily", + reducer=REDUCER, + window_rules=(metrics.CUMULATIVE_WEEKLY,), + ) + } + display = MetricDisplay( + friendly_name="Active hours", description="Time with input.", bigger_is_better=True + ) + experiment = experiment_with(["active_hours"], display={"active_hours": display}) + + chosen, _ = metrics.for_experiment(experiment, resolved, {}) + + assert chosen[0].friendly_name == "Active hours" + assert chosen[0].description == "Time with input." + + +def test_for_experiment_reports_a_name_it_asked_for_that_did_not_resolve(): + experiment = experiment_with(["an_outcome_metric"]) + + chosen, problems = metrics.for_experiment( + experiment, {}, {"an_outcome_metric": "no nimbus/ definition for this app"} + ) + + assert chosen == [] + assert len(problems) == 1 + assert problems[0].name == "an_outcome_metric" + assert "no nimbus/ definition" in problems[0].reason diff --git a/jobs/highwind/tests/test_output_writing.py b/jobs/highwind/tests/test_output_writing.py index f9994585f..5f4fad896 100644 --- a/jobs/highwind/tests/test_output_writing.py +++ b/jobs/highwind/tests/test_output_writing.py @@ -12,7 +12,7 @@ from highwind import discovery, gbstats_compute, output_writing, units from highwind.discovery import Experiment -from highwind.metrics import CUMULATIVE_WEEKLY, DISJOINT_WEEKLY, Metric, per_unit_sum +from highwind.metrics import CUMULATIVE_WEEKLY, DISJOINT_WEEKLY, Metric AS_OF = datetime.date(2026, 8, 1) @@ -404,7 +404,7 @@ def test_the_log_table_is_partitioned_and_clustered_like_the_tables_it_sits_besi CUMULATIVE_METRIC = Metric( name="active_hours", source="clients_daily", - reducer=per_unit_sum("active_hours_sum"), + reducer=None, window_rules=(CUMULATIVE_WEEKLY,), friendly_name="Active hours", description="Time with user input.", @@ -413,7 +413,7 @@ def test_the_log_table_is_partitioned_and_clustered_like_the_tables_it_sits_besi DISJOINT_METRIC = Metric( name="retained", source="clients_daily", - reducer=per_unit_sum("pings_aggregated_by_this_row"), + reducer=None, window_rules=(DISJOINT_WEEKLY,), friendly_name="Retained", ) diff --git a/jobs/highwind/tests/test_sql_generation.py b/jobs/highwind/tests/test_sql_generation.py index 400e53e14..a8214b37a 100644 --- a/jobs/highwind/tests/test_sql_generation.py +++ b/jobs/highwind/tests/test_sql_generation.py @@ -3,6 +3,8 @@ import datetime import pytest +from metric_config_parser.aggregation import BucketedAggregation, Combine +from metric_config_parser.aggregation_plan import AggregationSource from highwind import discovery, metrics, sql_generation, units from highwind.discovery import Experiment, Window @@ -13,6 +15,71 @@ CLIENT_UNIT = units.resolve("firefox_desktop", "normandy_id") GROUP_UNIT = units.resolve("firefox_desktop", "group_id") +CLIENTS_DAILY_SOURCE = AggregationSource( + data_source="clients_daily", + from_expression="mozdata.telemetry.clients_daily", + client_id_column="client_id", + group_id_column="profile_group_id", + analysis_units=("client_id", "profile_group_id"), + submission_date_column="submission_date", + columns=("active_hours_sum", "pings_aggregated_by_this_row"), +) +ACTIVE_USERS_SOURCE = AggregationSource( + data_source="active_users", + from_expression=( + "(\n SELECT * FROM mozdata.telemetry.desktop_active_users\n WHERE is_desktop\n)" + ), + client_id_column="client_id", + group_id_column="profile_group_id", + analysis_units=("client_id", "profile_group_id"), + submission_date_column="submission_date", + columns=("is_dau", "days_active_bits"), +) + + +def reducer(bucket_aggregate, combine=Combine.SUM, no_rows="0", finalize="{combined}"): + return BucketedAggregation( + bucket_aggregate=bucket_aggregate, combine=combine, no_rows=no_rows, finalize=finalize + ) + + +# The handful of desktop guardrail shapes these tests exercise, built directly rather than read off +# metric-hub: these tests are about sql_generation.py's own composition of a reducer into the bucket +# grid, not about what metric-hub happens to return today. +FIXTURE_METRICS = { + "active_hours": metrics.Metric( + name="active_hours", + source="clients_daily", + reducer=reducer("SUM(active_hours_sum)"), + window_rules=(metrics.CUMULATIVE_WEEKLY,), + ), + "retained": metrics.Metric( + name="retained", + source="clients_daily", + reducer=reducer( + "SUM(pings_aggregated_by_this_row)", finalize="CAST({combined} > 0 AS INT64)" + ), + window_rules=(metrics.DISJOINT_WEEKLY,), + ), + "retained_dau": metrics.Metric( + name="retained_dau", + source="active_users", + reducer=reducer("COUNTIF(is_dau)", finalize="CAST({combined} > 0 AS INT64)"), + window_rules=(metrics.DISJOINT_WEEKLY,), + ), + "active_in_last_3_days_legacy": metrics.Metric( + name="active_in_last_3_days_legacy", + source="active_users", + reducer=reducer( + "MIN(mozfun.bits28.days_since_seen(days_active_bits))", + combine=Combine.MIN, + no_rows="30", + finalize="CAST({combined} < 3 AS INT64)", + ), + window_rules=(metrics.DISJOINT_WEEKLY,), + ), +} + def experiment(slug, start_date, unit=CLIENT_UNIT): return Experiment( @@ -60,7 +127,7 @@ def run_of(windows_by_slug, sample_percent=None, units_by_slug=None): def named(name): - return next(metric for metric in metrics.GUARDRAILS if metric.name == name) + return FIXTURE_METRICS[name] def test_a_weekly_window_set_yields_a_weekly_bucket_grid(): @@ -182,7 +249,12 @@ def test_a_metric_that_combines_with_min_takes_a_running_minimum(): metric = metrics.Metric( name="a-metric", source="active_users", - reducer=metrics.per_unit_min_below("days_since_seen", 3, 30), + reducer=reducer( + "MIN(days_since_seen)", + combine=Combine.MIN, + no_rows="30", + finalize="CAST({combined} < 3 AS INT64)", + ), window_rules=(metrics.CUMULATIVE_WEEKLY,), ) value = sql_generation.window_value(metric, "cumulative") @@ -237,7 +309,7 @@ def test_a_client_randomized_run_samples_the_sources_on_the_clustered_column(): cohort = sql_generation.cohort_query(run) source = sql_generation.source_cte( - sql_generation.SOURCES["clients_daily"], run, CLIENT_UNIT + CLIENTS_DAILY_SOURCE, run, CLIENT_UNIT ) assert "udf.safe_sample_id(unit_id) < 1" in cohort @@ -254,7 +326,7 @@ def test_a_group_randomized_run_samples_the_sources_on_the_group_rather_than_the ) source = sql_generation.source_cte( - sql_generation.SOURCES["clients_daily"], run, GROUP_UNIT + CLIENTS_DAILY_SOURCE, run, GROUP_UNIT ) assert "udf.safe_sample_id(profile_group_id) < 1" in source @@ -267,7 +339,7 @@ def test_an_unsampled_run_restricts_neither_the_cohort_nor_the_sources(): assert "safe_sample_id" not in sql_generation.cohort_query(run) for unit in (CLIENT_UNIT, GROUP_UNIT): source = sql_generation.source_cte( - sql_generation.SOURCES["clients_daily"], run, unit + CLIENTS_DAILY_SOURCE, run, unit ) assert "sample_id <" not in source @@ -287,7 +359,7 @@ def test_the_sql_and_discovery_build_a_window_label_the_same_way(): metric = metrics.Metric( name="probe", source="clients_daily", - reducer=metrics.per_unit_sum("active_hours_sum"), + reducer=None, window_rules=(rule,), ) windows = discovery.windows_for(metric, tenure_days=60) @@ -301,7 +373,7 @@ def test_the_sql_and_discovery_build_a_window_label_the_same_way(): def test_a_source_query_reads_the_column_the_experiments_unit_names(): run = run_of({"a-slug": cumulative_windows(2)}) - source_definition = sql_generation.SOURCES["clients_daily"] + source_definition = CLIENTS_DAILY_SOURCE assert "profile_group_id AS unit_id" in sql_generation.source_cte( source_definition, run, GROUP_UNIT @@ -414,7 +486,8 @@ def test_a_run_is_split_into_one_query_per_source_and_unit(): "clients_daily": [named("active_hours")], "active_users": [named("retained_dau")], } - queries = sql_generation.build_queries(run, metrics_by_source, "a.cohort_table") + sources = {"clients_daily": CLIENTS_DAILY_SOURCE, "active_users": ACTIVE_USERS_SOURCE} + queries = sql_generation.build_queries(run, metrics_by_source, sources, "a.cohort_table") assert set(queries) == { "clients_daily/client_id", @@ -429,7 +502,10 @@ def test_a_run_is_split_into_one_query_per_source_and_unit(): def test_a_run_whose_experiments_all_randomize_alike_stays_one_query_per_source(): run = run_of({"a-slug": cumulative_windows(2), "b-slug": cumulative_windows(2)}) queries = sql_generation.build_queries( - run, {"clients_daily": [named("active_hours")]}, "a.cohort_table" + run, + {"clients_daily": [named("active_hours")]}, + {"clients_daily": CLIENTS_DAILY_SOURCE}, + "a.cohort_table", ) assert set(queries) == {"clients_daily/client_id"} diff --git a/jobs/highwind/tests/test_units.py b/jobs/highwind/tests/test_units.py index ceaa37c74..c1381df60 100644 --- a/jobs/highwind/tests/test_units.py +++ b/jobs/highwind/tests/test_units.py @@ -1,6 +1,18 @@ """Resolving a recipe's analysis unit, and refusing to invent one.""" -from highwind import metrics, units +from metric_config_parser.aggregation_plan import AggregationSource + +from highwind import units + +CLIENTS_DAILY = AggregationSource( + data_source="clients_daily", + from_expression="mozdata.telemetry.clients_daily", + client_id_column="client_id", + group_id_column="profile_group_id", + analysis_units=("client_id", "profile_group_id"), + submission_date_column="submission_date", + columns=("active_hours_sum",), +) def test_a_group_randomized_desktop_recipe_is_analysed_at_the_profile_group(): @@ -8,7 +20,7 @@ def test_a_group_randomized_desktop_recipe_is_analysed_at_the_profile_group(): assert unit.kind == units.PROFILE_GROUP_ID assert unit.enrollment_column == "profile_group_id" - assert unit.source_column == "profile_group_id" + assert unit.source_column(CLIENTS_DAILY) == "profile_group_id" def test_the_desktop_client_units_all_resolve_to_the_same_unit(): @@ -24,7 +36,7 @@ def test_the_enrollment_and_source_columns_differ_for_a_client_randomized_recipe unit = units.resolve("firefox_desktop", "normandy_id") assert unit.enrollment_column == "legacy_telemetry_client_id" - assert unit.source_column == "client_id" + assert unit.source_column(CLIENTS_DAILY) == "client_id" def test_only_the_client_unit_matches_the_clustered_sample_column(): @@ -51,7 +63,7 @@ def test_no_unit_column_is_also_a_metric_column(): # The source CTE selects the unit column aliased to unit_id and then the metric columns, so a # column named by both would be selected twice under two names. for unit in set(units.ANALYSIS_UNITS.values()): - assert unit.source_column + column = unit.source_column(CLIENTS_DAILY) + assert column assert unit.enrollment_column - for spec in metrics.SOURCES.values(): - assert unit.source_column not in spec["columns"] + assert column not in CLIENTS_DAILY.columns From b61063bc98477f288a9b787baafaec9d2042232d Mon Sep 17 00:00:00 2001 From: Jared Lockhart <119884+jaredlockhart@users.noreply.github.com> Date: Tue, 27 Oct 2026 17:29:34 -0400 Subject: [PATCH 2/3] fix(highwind): report unsupported windows and skip unneeded source scans Because * A metric-hub window that doesn't fit the bucket grid stopped the whole run * Every source was scanned for every unit with every experiment's metrics This commit * Reports metrics whose windows don't fit the 7-day bucket grid instead of failing the run * Builds each unit's queries from its own experiments' metrics and skips source and unit pairs with nothing to compute https://mozilla-hub.atlassian.net/browse/EXP-7631 --- jobs/highwind/highwind/analysis.py | 18 +++--- jobs/highwind/highwind/metrics.py | 55 ++++++++++++----- jobs/highwind/highwind/sql_generation.py | 32 ++++++---- jobs/highwind/tests/test_analysis.py | 11 +++- jobs/highwind/tests/test_metrics.py | 38 +++++++++++- jobs/highwind/tests/test_sql_generation.py | 69 +++++++++++++++++++--- 6 files changed, 176 insertions(+), 47 deletions(-) diff --git a/jobs/highwind/highwind/analysis.py b/jobs/highwind/highwind/analysis.py index 379c0681f..aca651e0e 100644 --- a/jobs/highwind/highwind/analysis.py +++ b/jobs/highwind/highwind/analysis.py @@ -87,7 +87,7 @@ def run_daily_job( if not experiments: return [] - metrics_by_slug, metrics_by_source, sources = resolve_run_metrics(experiments) + metrics_by_slug, resolved_metrics, sources = resolve_run_metrics(experiments) windows_by_slug = { experiment.slug: run_windows(experiment, as_of, metrics_by_slug[experiment.slug]) for experiment in experiments @@ -97,7 +97,7 @@ def run_daily_job( ) cells_by_slug, units_by_slug, timings, failure = gather_sufficient_statistics( - client, run, metrics_by_source, sources, 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 @@ -317,8 +317,9 @@ def writes_blobs(validate_only, sample_percent, outputs): 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 grouped by source (for the shared per-source scan), and the plan's source + 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. @@ -338,11 +339,12 @@ def resolve_run_metrics(experiments): f"{experiment.slug}: metric {problem.name!r} not computed: {problem.reason}", extra={"experiment_slug": experiment.slug, "metric": problem.name}, ) - return metrics_by_slug, metrics.group_by_source(resolved_metrics), plan.sources + return metrics_by_slug, resolved_metrics, plan.sources -def gather_sufficient_statistics(client, run, metrics_by_source, sources, as_of, validate_only): - """Run the shared scan: cohort once, then one query per source and unit, split by slug. +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 @@ -354,7 +356,7 @@ def gather_sufficient_statistics(client, run, metrics_by_source, sources, as_of, 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, sources, 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 ) diff --git a/jobs/highwind/highwind/metrics.py b/jobs/highwind/highwind/metrics.py index c7da8cd7c..3e0a66650 100644 --- a/jobs/highwind/highwind/metrics.py +++ b/jobs/highwind/highwind/metrics.py @@ -12,7 +12,11 @@ counts, carried over as Highwind's own cumulative and disjoint window rules. Only a *repeating* family has a Highwind equivalent (see `sql_generation.py`'s window-rule docstring), so a metric whose `repeat_windows` is false, or that declares neither window, has no window rule Highwind can -build and is reported unresolved rather than silently dropped or mis-rendered. +build and is reported unresolved rather than silently dropped or mis-rendered. The same holds for +a window that does not fit Highwind's own bucket grid (`BUCKET_LENGTH`): an `incremental_window` +must be exactly one bucket and a `cumulative_window` an exact multiple of one, or there is no grid +to read it off, and reporting it here is what keeps one metric's misfit window from failing the +whole run (see `sql_generation.bucket_length`). """ from dataclasses import dataclass @@ -25,11 +29,18 @@ CUMULATIVE = "cumulative" DISJOINT = "disjoint" +# Highwind's own bucket grid (see `sql_generation.py`'s `Run` and `bucket_length`): every window a +# run builds is read off buckets this many days long, no matter which metrics the run happens to +# carry. Fixed here, at resolution time, rather than derived from the windows a run's metrics +# declare, so a metric whose window does not fit it is reported unresolved instead of changing the +# grid every other metric in the run is read against. +BUCKET_LENGTH = 7 + # The only lengths metric-hub's desktop guardrails declare today. Kept as named fixtures because # every window rule built from the plan is one of these two shapes; nothing downstream depends on # the names themselves. -CUMULATIVE_WEEKLY = {"kind": CUMULATIVE, "length": 7} -DISJOINT_WEEKLY = {"kind": DISJOINT, "length": 7} +CUMULATIVE_WEEKLY = {"kind": CUMULATIVE, "length": BUCKET_LENGTH} +DISJOINT_WEEKLY = {"kind": DISJOINT, "length": BUCKET_LENGTH} @dataclass(frozen=True) @@ -71,13 +82,17 @@ def load_plan(configs, metric_names): def window_rules(aggregation_metric): """The window rules Highwind can build from a plan metric's window fields, or why not. - Returns `(rules, None)` when the metric has at least one window family and repeats it, or - `(None, reason)` otherwise. A metric with neither `cumulative_window` nor `incremental_window` - set has nothing for Highwind to compute a window over at all; one that sets `repeat_windows` - false declares a single window that stops, which none of Highwind's window rules can express, - since every rule it has ever supported extends for as long as the experiment runs (see - `sql_generation.py`'s `Run` docstring). Mapping either case onto a repeating rule would compute - windows the metric never declared. + Returns `(rules, None)` when the metric has at least one window family, repeats it, and that + family fits Highwind's bucket grid, or `(None, reason)` otherwise. A metric with neither + `cumulative_window` nor `incremental_window` set has nothing for Highwind to compute a window + over at all; one that sets `repeat_windows` false declares a single window that stops, which + none of Highwind's window rules can express, since every rule it has ever supported extends for + as long as the experiment runs (see `sql_generation.py`'s `Run` docstring). An + `incremental_window` that is not exactly one `BUCKET_LENGTH`, or a `cumulative_window` that is + not an exact multiple of one, has no grid to read off: a disjoint window must BE one bucket and + a cumulative one is summed from whole buckets (see `sql_generation.bucket_length`'s own + backstop validation). Mapping any of these cases onto a repeating rule would compute windows + the metric never declared, or on a grid that does not fit. """ no_window = ( aggregation_metric.cumulative_window is None @@ -90,11 +105,23 @@ def window_rules(aggregation_metric): "repeat_windows is false; Highwind's window rules always repeat and cannot " "express a one-off window" ) + incremental = aggregation_metric.incremental_window + if incremental is not None and incremental != BUCKET_LENGTH: + return None, ( + f"incremental_window is {incremental} days, not {BUCKET_LENGTH}; a disjoint window " + f"must be exactly one bucket on Highwind's {BUCKET_LENGTH}-day grid" + ) + cumulative = aggregation_metric.cumulative_window + if cumulative is not None and cumulative % BUCKET_LENGTH: + return None, ( + f"cumulative_window is {cumulative} days, not a multiple of Highwind's " + f"{BUCKET_LENGTH}-day bucket grid" + ) rules = [] - if aggregation_metric.cumulative_window is not None: - rules.append({"kind": CUMULATIVE, "length": aggregation_metric.cumulative_window}) - if aggregation_metric.incremental_window is not None: - rules.append({"kind": DISJOINT, "length": aggregation_metric.incremental_window}) + if cumulative is not None: + rules.append({"kind": CUMULATIVE, "length": cumulative}) + if incremental is not None: + rules.append({"kind": DISJOINT, "length": incremental}) return tuple(rules), None diff --git a/jobs/highwind/highwind/sql_generation.py b/jobs/highwind/highwind/sql_generation.py index c78b7ea51..48e013b0f 100644 --- a/jobs/highwind/highwind/sql_generation.py +++ b/jobs/highwind/highwind/sql_generation.py @@ -44,6 +44,7 @@ import math import textwrap +from . import metrics from .units import PROFILE_GROUP_ID # The bucket a unit's pre-enrollment rows land in. The covariate is one fixed window rather than @@ -452,20 +453,29 @@ def enrollment_unit_id(run): # ----------------------------------------------------------------------- the source queries ---- -def build_queries(run, metrics_by_source, sources, cohort_table): - """One SQL string per (source table, analysis unit), covering every experiment in the run. +def build_queries(run, resolved_metrics, sources, cohort_table): + """One SQL string per (source table, analysis unit) that has something to compute. Keyed by both, because a run whose experiments randomize on different units reads each source - once per unit and the two are separate queries with separate costs. `sources` is the - aggregation plan's resolved source definitions, keyed the same way as `metrics_by_source`. + once per unit and the two are separate queries with separate costs. Scoped to each unit's own + experiments: a source whose metrics only ever appear at a different unit contributes no group + for this one, so that (source, unit) pair is skipped rather than scanned for metrics nothing + there reads. `sources` is the aggregation plan's resolved source definitions, keyed by the same + source names as `resolved_metrics`' own `.source` field. """ - return { - f"{source}/{unit.kind}": build_source_query( - unit_run, source, metrics, sources[source], cohort_table, unit - ) - for unit, unit_run in run.by_unit().items() - for source, metrics in metrics_by_source.items() - } + queries = {} + for unit, unit_run in run.by_unit().items(): + names = { + name for experiment in unit_run.experiments for name in experiment.metric_names + } + unit_metrics = { + name: metric for name, metric in resolved_metrics.items() if name in names + } + for source, source_metrics in metrics.group_by_source(unit_metrics).items(): + queries[f"{source}/{unit.kind}"] = build_source_query( + unit_run, source, source_metrics, sources[source], cohort_table, unit + ) + return queries def build_source_query(run, source, metrics, source_definition, cohort_table, unit): diff --git a/jobs/highwind/tests/test_analysis.py b/jobs/highwind/tests/test_analysis.py index e0244f75d..aa99ee76b 100644 --- a/jobs/highwind/tests/test_analysis.py +++ b/jobs/highwind/tests/test_analysis.py @@ -10,7 +10,7 @@ from mozilla_nimbus_schemas.highwind import HighwindAnalysis from highwind import analysis, output_writing, sql_generation, units -from highwind.discovery import Experiment +from highwind.discovery import Experiment, MetricDisplay from highwind.gbstats_compute import ( CONFIDENT, ERROR, @@ -29,6 +29,11 @@ reference_branch="control", treatment_branches=("treatment-a",), unit=units.resolve("firefox_desktop", "normandy_id"), + metrics={ + "active_hours": MetricDisplay( + friendly_name=None, description=None, bigger_is_better=None + ) + }, ) # One resolved metric, standing in for what `metrics.resolve()` would build off a real aggregation @@ -500,7 +505,7 @@ def update_table(self, table, fields): def test_a_failed_unit_count_keeps_the_cells_the_shared_scan_computed(caplog): # The unit counts are display-only, so losing them must not turn every experiment's computed # statistics into an error grid. - metrics_by_source = {"clients_daily": TEST_METRICS} + resolved_metrics = {"active_hours": TEST_METRICS[0]} run = sql_generation.Run( [EXPERIMENT], {"a-slug": analysis.run_windows(EXPERIMENT, AS_OF, TEST_METRICS)}, @@ -510,7 +515,7 @@ def test_a_failed_unit_count_keeps_the_cells_the_shared_scan_computed(caplog): cells_by_slug, units_by_slug, timings, failure = ( analysis.gather_sufficient_statistics( - UnitCountFailingClient(), run, metrics_by_source, TEST_SOURCES, AS_OF, False + UnitCountFailingClient(), run, resolved_metrics, TEST_SOURCES, AS_OF, False ) ) diff --git a/jobs/highwind/tests/test_metrics.py b/jobs/highwind/tests/test_metrics.py index dc982ea18..9bb33253a 100644 --- a/jobs/highwind/tests/test_metrics.py +++ b/jobs/highwind/tests/test_metrics.py @@ -53,12 +53,12 @@ def test_an_incremental_window_becomes_a_disjoint_rule(): def test_both_windows_together_become_both_rules(): rules, reason = metrics.window_rules( - aggregation_metric(cumulative_window=7, incremental_window=3) + aggregation_metric(cumulative_window=14, incremental_window=7) ) assert rules == ( - {"kind": "cumulative", "length": 7}, - {"kind": "disjoint", "length": 3}, + {"kind": "cumulative", "length": 14}, + {"kind": "disjoint", "length": 7}, ) assert reason is None @@ -81,6 +81,38 @@ def test_a_non_repeating_window_is_unsupported(): assert "repeat_windows" in reason +def test_an_incremental_window_that_is_not_one_bucket_is_reported_not_raised(): + rules, reason = metrics.window_rules( + aggregation_metric(cumulative_window=None, incremental_window=14) + ) + + assert rules is None + assert "incremental_window" in reason + + +def test_a_cumulative_window_that_is_not_a_multiple_of_the_bucket_is_reported(): + rules, reason = metrics.window_rules(aggregation_metric(cumulative_window=10)) + + assert rules is None + assert "cumulative_window" in reason + + +def test_a_cumulative_window_that_is_a_multiple_of_the_bucket_is_accepted(): + rules, reason = metrics.window_rules(aggregation_metric(cumulative_window=14)) + + assert rules == ({"kind": "cumulative", "length": 14},) + assert reason is None + + +def test_resolve_reports_a_misfit_window_the_same_way_as_repeat_windows_false(): + plan = plan_of({"active_hours": aggregation_metric(cumulative_window=10)}) + + resolved, unresolved = metrics.resolve(plan) + + assert resolved == {} + assert "cumulative_window" in unresolved["active_hours"] + + def test_resolve_splits_missing_names_into_unresolved_with_a_reason(): plan = plan_of({}, missing=["no_such_metric"]) diff --git a/jobs/highwind/tests/test_sql_generation.py b/jobs/highwind/tests/test_sql_generation.py index a8214b37a..75ecd32a5 100644 --- a/jobs/highwind/tests/test_sql_generation.py +++ b/jobs/highwind/tests/test_sql_generation.py @@ -81,7 +81,7 @@ def reducer(bucket_aggregate, combine=Combine.SUM, no_rows="0", finalize="{combi } -def experiment(slug, start_date, unit=CLIENT_UNIT): +def experiment(slug, start_date, unit=CLIENT_UNIT, metric_names=()): return Experiment( slug=slug, start_date=start_date, @@ -89,6 +89,12 @@ def experiment(slug, start_date, unit=CLIENT_UNIT): reference_branch="control", treatment_branches=("treatment",), unit=unit, + metrics={ + name: discovery.MetricDisplay( + friendly_name=None, description=None, bigger_is_better=None + ) + for name in metric_names + }, ) @@ -111,13 +117,15 @@ def disjoint_windows(count): ] -def run_of(windows_by_slug, sample_percent=None, units_by_slug=None): +def run_of(windows_by_slug, sample_percent=None, units_by_slug=None, metric_names_by_slug=None): units_by_slug = units_by_slug or {} + metric_names_by_slug = metric_names_by_slug or {} experiments = [ experiment( slug, AS_OF - datetime.timedelta(days=60), units_by_slug.get(slug, CLIENT_UNIT), + metric_names_by_slug.get(slug, ()), ) for slug in windows_by_slug ] @@ -481,13 +489,17 @@ def test_a_run_is_split_into_one_query_per_source_and_unit(): run = run_of( {"grouped": cumulative_windows(2), "by-client": cumulative_windows(2)}, units_by_slug={"grouped": GROUP_UNIT}, + metric_names_by_slug={ + "grouped": ("active_hours", "retained_dau"), + "by-client": ("active_hours", "retained_dau"), + }, ) - metrics_by_source = { - "clients_daily": [named("active_hours")], - "active_users": [named("retained_dau")], + resolved_metrics = { + "active_hours": named("active_hours"), + "retained_dau": named("retained_dau"), } sources = {"clients_daily": CLIENTS_DAILY_SOURCE, "active_users": ACTIVE_USERS_SOURCE} - queries = sql_generation.build_queries(run, metrics_by_source, sources, "a.cohort_table") + queries = sql_generation.build_queries(run, resolved_metrics, sources, "a.cohort_table") assert set(queries) == { "clients_daily/client_id", @@ -500,10 +512,13 @@ def test_a_run_is_split_into_one_query_per_source_and_unit(): def test_a_run_whose_experiments_all_randomize_alike_stays_one_query_per_source(): - run = run_of({"a-slug": cumulative_windows(2), "b-slug": cumulative_windows(2)}) + run = run_of( + {"a-slug": cumulative_windows(2), "b-slug": cumulative_windows(2)}, + metric_names_by_slug={"a-slug": ("active_hours",), "b-slug": ("active_hours",)}, + ) queries = sql_generation.build_queries( run, - {"clients_daily": [named("active_hours")]}, + {"active_hours": named("active_hours")}, {"clients_daily": CLIENTS_DAILY_SOURCE}, "a.cohort_table", ) @@ -511,6 +526,44 @@ def test_a_run_whose_experiments_all_randomize_alike_stays_one_query_per_source( assert set(queries) == {"clients_daily/client_id"} +def test_a_source_used_only_at_one_unit_builds_no_query_for_the_other(): + run = run_of( + {"grouped": cumulative_windows(2), "by-client": cumulative_windows(2)}, + units_by_slug={"grouped": GROUP_UNIT}, + metric_names_by_slug={ + "grouped": ("retained_dau",), + "by-client": ("active_hours",), + }, + ) + resolved_metrics = { + "active_hours": named("active_hours"), + "retained_dau": named("retained_dau"), + } + sources = {"clients_daily": CLIENTS_DAILY_SOURCE, "active_users": ACTIVE_USERS_SOURCE} + + queries = sql_generation.build_queries(run, resolved_metrics, sources, "a.cohort_table") + + assert set(queries) == {"active_users/profile_group_id", "clients_daily/client_id"} + + +def test_each_units_query_carries_only_its_own_experiments_metrics(): + run = run_of( + {"grouped": cumulative_windows(2), "by-client": cumulative_windows(2)}, + units_by_slug={"grouped": GROUP_UNIT}, + metric_names_by_slug={ + "grouped": ("active_hours",), + "by-client": ("active_hours", "retained"), + }, + ) + resolved_metrics = {"active_hours": named("active_hours"), "retained": named("retained")} + sources = {"clients_daily": CLIENTS_DAILY_SOURCE} + + queries = sql_generation.build_queries(run, resolved_metrics, sources, "a.cohort_table") + + assert "retained__raw" not in queries["clients_daily/profile_group_id"] + assert "retained__raw" in queries["clients_daily/client_id"] + + def test_each_units_source_scan_reaches_back_only_as_far_as_its_own_experiments(): # The sub-runs are what make this true: one unit's oldest experiment must not widen the other's # scan, which it would if every source query were built from the whole run. From 8cffdde39c8615c3e2eaaad979d0254b4cdfcbf6 Mon Sep 17 00:00:00 2001 From: Jared Lockhart <119884+jaredlockhart@users.noreply.github.com> Date: Wed, 28 Oct 2026 17:13:39 -0400 Subject: [PATCH 3/3] refactor(highwind): address review of metric-hub loading Because * Calls through a module named metrics read like operations on a list of metrics * The discovery query grouped a list that is already one entry per metric This commit * Imports the metric-hub loading functions by name, with names that say what each does * Reads each experiment's metric list straight from the mirror without regrouping it * Rewords the run_windows docstring https://mozilla-hub.atlassian.net/browse/EXP-7631 --- jobs/highwind/highwind/analysis.py | 20 ++++++++++++++------ jobs/highwind/highwind/discovery.py | 21 ++++++++++----------- jobs/highwind/highwind/metrics.py | 8 ++++---- jobs/highwind/tests/test_analysis.py | 4 ++-- jobs/highwind/tests/test_discovery.py | 5 +++-- jobs/highwind/tests/test_metrics.py | 14 +++++++------- 6 files changed, 40 insertions(+), 32 deletions(-) diff --git a/jobs/highwind/highwind/analysis.py b/jobs/highwind/highwind/analysis.py index aca651e0e..b5253c903 100644 --- a/jobs/highwind/highwind/analysis.py +++ b/jobs/highwind/highwind/analysis.py @@ -25,8 +25,14 @@ from google.cloud import storage # type: ignore from google.cloud import bigquery -from . import discovery, gbstats_compute, metrics +from . import discovery, gbstats_compute 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__) @@ -324,13 +330,13 @@ def resolve_run_metrics(experiments): `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 = metrics.load_configs() + configs = load_metric_hub() requested = {name for experiment in experiments for name in experiment.metric_names} - plan = metrics.load_plan(configs, requested) - resolved_metrics, unresolved_metrics = metrics.resolve(plan) + 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( + exp_metrics, problems = metrics_for_experiment( experiment, resolved_metrics, unresolved_metrics ) metrics_by_slug[experiment.slug] = exp_metrics @@ -469,7 +475,9 @@ def failed_summary(experiment, windows, metrics, error): 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, out to the maturity frontier. + `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) seen, windows = set(), [] diff --git a/jobs/highwind/highwind/discovery.py b/jobs/highwind/highwind/discovery.py index 2612875cd..2496cb9d2 100644 --- a/jobs/highwind/highwind/discovery.py +++ b/jobs/highwind/highwind/discovery.py @@ -61,16 +61,13 @@ -- 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. - -- Grouped by name because the mirror carries one row per (metric, statistic), so the same metric - -- name can repeat; ANY_VALUE picks the metric's own display text off whichever row survives. ARRAY( SELECT AS STRUCT metric.name, - ANY_VALUE(metric.friendly_name) AS friendly_name, - ANY_VALUE(metric.description) AS description, - ANY_VALUE(metric.bigger_is_better) AS bigger_is_better + metric.friendly_name, + metric.description, + metric.bigger_is_better FROM UNNEST(metric_config.metrics) AS metric - GROUP BY metric.name ) AS metric_configs FROM ( SELECT @@ -148,8 +145,9 @@ 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 @@ -157,8 +155,8 @@ def tenure_days(self, as_of): def metric_display(metric_configs): """Turn the mirror's per-experiment metric rows into a name-keyed display map. - The mirror has already deduplicated by name (see `DISCOVERY_SQL`'s `GROUP BY metric.name`), so - this is a straight rekey rather than a second dedup. + `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( @@ -258,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 diff --git a/jobs/highwind/highwind/metrics.py b/jobs/highwind/highwind/metrics.py index 3e0a66650..086818668 100644 --- a/jobs/highwind/highwind/metrics.py +++ b/jobs/highwind/highwind/metrics.py @@ -64,7 +64,7 @@ class UnresolvedMetric: reason: str -def load_configs(): +def load_metric_hub(): """Load metric-hub's own `nimbus/` definitions, once per run. A fresh clone of `mozilla/metric-hub` every call (see metric-config-parser's @@ -74,7 +74,7 @@ def load_configs(): return ConfigCollection.from_github_repo() -def load_plan(configs, metric_names): +def build_aggregation_plan(configs, metric_names): """Resolve one run's metric names against metric-hub's `nimbus/` definitions for this app.""" return aggregation_plan(configs, APP_NAME, sorted(set(metric_names))) @@ -125,7 +125,7 @@ def window_rules(aggregation_metric): return tuple(rules), None -def resolve(plan): +def split_resolved_metrics(plan): """Split one run's plan into the metrics Highwind can compute and the ones it cannot, with why. A name fails to resolve two ways: no `nimbus/` definition at all (`plan.missing`), or a @@ -162,7 +162,7 @@ def group_by_source(resolved_metrics): return by_source -def for_experiment(experiment, resolved_metrics, unresolved_metrics): +def metrics_for_experiment(experiment, resolved_metrics, unresolved_metrics): """One experiment's own metric list, with its own display text, and the names it could not get. A metric's aggregation shape is the same for every experiment that reads it, so diff --git a/jobs/highwind/tests/test_analysis.py b/jobs/highwind/tests/test_analysis.py index aa99ee76b..c4ac22e3a 100644 --- a/jobs/highwind/tests/test_analysis.py +++ b/jobs/highwind/tests/test_analysis.py @@ -36,8 +36,8 @@ }, ) -# One resolved metric, standing in for what `metrics.resolve()` would build off a real aggregation -# plan: these tests exercise the run's orchestration, not metric resolution itself (see +# One resolved metric, standing in for what `split_resolved_metrics()` would build off a real +# aggregation plan: these tests exercise the run's orchestration, not metric resolution itself (see # test_metrics.py for that). TEST_METRICS = [ Metric( diff --git a/jobs/highwind/tests/test_discovery.py b/jobs/highwind/tests/test_discovery.py index 868f5dd83..c20462590 100644 --- a/jobs/highwind/tests/test_discovery.py +++ b/jobs/highwind/tests/test_discovery.py @@ -164,8 +164,9 @@ def test_an_experiment_carries_its_own_deduplicated_metric_names(): mirror_row( metric_configs=[ metric_config("active_hours", friendly_name="Active hours"), - # The mirror carries one row per (metric, statistic), so the same metric - # name can repeat; discover() reads the dedup DISCOVERY_SQL already did. + # `metric_config.metrics` holds one entry per metric, so a repeated name is + # not expected from the mirror; `metric_display` keys on name regardless, so + # a repeat still collapses to one entry rather than raising. metric_config("active_hours", friendly_name="Active hours"), metric_config("retained_dau", friendly_name="Retained (DAU)"), ] diff --git a/jobs/highwind/tests/test_metrics.py b/jobs/highwind/tests/test_metrics.py index 9bb33253a..8c7c0043e 100644 --- a/jobs/highwind/tests/test_metrics.py +++ b/jobs/highwind/tests/test_metrics.py @@ -107,7 +107,7 @@ def test_a_cumulative_window_that_is_a_multiple_of_the_bucket_is_accepted(): def test_resolve_reports_a_misfit_window_the_same_way_as_repeat_windows_false(): plan = plan_of({"active_hours": aggregation_metric(cumulative_window=10)}) - resolved, unresolved = metrics.resolve(plan) + resolved, unresolved = metrics.split_resolved_metrics(plan) assert resolved == {} assert "cumulative_window" in unresolved["active_hours"] @@ -116,7 +116,7 @@ def test_resolve_reports_a_misfit_window_the_same_way_as_repeat_windows_false(): def test_resolve_splits_missing_names_into_unresolved_with_a_reason(): plan = plan_of({}, missing=["no_such_metric"]) - resolved, unresolved = metrics.resolve(plan) + resolved, unresolved = metrics.split_resolved_metrics(plan) assert resolved == {} assert "no nimbus/ definition" in unresolved["no_such_metric"] @@ -125,7 +125,7 @@ def test_resolve_splits_missing_names_into_unresolved_with_a_reason(): def test_resolve_reports_an_unsupported_window_the_same_way_as_a_missing_name(): plan = plan_of({"active_hours": aggregation_metric(repeat_windows=False)}) - resolved, unresolved = metrics.resolve(plan) + resolved, unresolved = metrics.split_resolved_metrics(plan) assert resolved == {} assert "repeat_windows" in unresolved["active_hours"] @@ -134,7 +134,7 @@ def test_resolve_reports_an_unsupported_window_the_same_way_as_a_missing_name(): def test_resolve_builds_a_metric_from_a_resolvable_plan_entry(): plan = plan_of({"active_hours": aggregation_metric()}) - resolved, unresolved = metrics.resolve(plan) + resolved, unresolved = metrics.split_resolved_metrics(plan) assert unresolved == {} metric = resolved["active_hours"] @@ -211,7 +211,7 @@ def test_for_experiment_keeps_only_the_names_it_asked_for(): } experiment = experiment_with(["active_hours"]) - chosen, problems = metrics.for_experiment(experiment, resolved, {}) + chosen, problems = metrics.metrics_for_experiment(experiment, resolved, {}) assert [metric.name for metric in chosen] == ["active_hours"] assert problems == [] @@ -231,7 +231,7 @@ def test_for_experiment_attaches_its_own_display_text(): ) experiment = experiment_with(["active_hours"], display={"active_hours": display}) - chosen, _ = metrics.for_experiment(experiment, resolved, {}) + chosen, _ = metrics.metrics_for_experiment(experiment, resolved, {}) assert chosen[0].friendly_name == "Active hours" assert chosen[0].description == "Time with input." @@ -240,7 +240,7 @@ def test_for_experiment_attaches_its_own_display_text(): def test_for_experiment_reports_a_name_it_asked_for_that_did_not_resolve(): experiment = experiment_with(["an_outcome_metric"]) - chosen, problems = metrics.for_experiment( + chosen, problems = metrics.metrics_for_experiment( experiment, {}, {"an_outcome_metric": "no nimbus/ definition for this app"} )