Data Mesh in Practice — Domain Ownership, Data Products, and Federated Governance

The Problem with Centralized Data Teams

The central data team bottleneck is one of the most predictable failure modes in data organizations. A single platform team owns the data warehouse, the pipelines, and the transformation logic for the entire company. Every product team that wants a new dashboard, a new data feed, or a schema change must file a ticket and wait. The data team becomes a dependency, the queue grows, and data consumers lose trust in the timeliness and accuracy of what they receive.

Data Mesh, articulated by Zhamak Dehghani at ThoughtWorks, addresses this by treating data as a product, distributing ownership to the domains that generate it, and providing a self-serve platform so domain teams can operate independently. The four principles — domain ownership, data as a product, self-serve data platform, and federated computational governance — work as a system. Adopting one without the others produces an incomplete architecture that creates new problems while solving old ones. The foundational concepts are covered in our Data Mesh Architecture overview; this article focuses on the implementation mechanics.

# The symptom: central data team ticket queue growing faster than capacity
# Measured at a 200-person tech company before Data Mesh adoption:

central_team_metrics = {
    "engineers": 6,
    "open_tickets": 142,
    "avg_ticket_age_days": 18,
    "pipeline_count": 340,
    "pipelines_with_no_documented_owner": 0.71,  # 71% unowned
    "data_incidents_per_month": 23,
    "mean_time_to_detect_hours": 14.2,
    "mean_time_to_resolve_hours": 31.5,
}

# The root cause: ownership mismatch
# Orders data is owned by the Orders team in production but by the Data team in analytics.
# When the Orders team changes the database schema, the pipeline breaks silently.
# The Data team learns about it from an angry business stakeholder two weeks later.

# Data Mesh solution: the Orders domain team owns both — the operational system AND
# the analytical data product derived from it. They know when schemas change. They fix
# their own data product. The central team governs standards, not pipelines.

Domain Identification — Drawing the Right Boundaries

Domain boundaries in Data Mesh should follow the same lines as your organizational domains, which in turn should follow bounded contexts in Domain-Driven Design. A domain is a cohesive business capability with a clear owner: Orders, Inventory, Customer, Payments, Marketing, Fraud. The team responsible for the operational system becomes the owner of the analytical data products derived from it. The key test is: when a fact changes in the real world, which team is the first to know? That team owns the data.

Domains are not tables or systems — they are business capabilities. The "Customer" domain may own data from a CRM, an identity service, and a billing platform, because customer identity is a single business concept even if it spans multiple technical systems. Conversely, splitting "Orders" into "Order Creation" and "Order Fulfillment" as separate domains is usually wrong — fulfillment is internal to the orders capability, not a separate business domain.

# Domain discovery workshop output — mapping capabilities to potential domains

domain_map = {
    "orders": {
        "team": "Order Experience Squad",
        "source_systems": ["orders_service_postgres", "returns_service_postgres"],
        "business_events": ["order_placed", "order_cancelled", "return_initiated", "return_resolved"],
        "analytical_questions_owned": [
            "What is our order conversion rate by channel?",
            "What is the refund rate by product category?",
            "What is the average order value trend?",
        ],
        "consumers": ["finance", "marketing", "supply_chain", "executive_reporting"],
    },
    "inventory": {
        "team": "Supply Chain Engineering",
        "source_systems": ["wms_oracle", "supplier_api_kafka_topic"],
        "business_events": ["stock_received", "stock_allocated", "stock_depleted", "reorder_triggered"],
        "analytical_questions_owned": [
            "What is our current stock level by SKU and warehouse?",
            "Which SKUs are at risk of stockout in the next 14 days?",
        ],
        "consumers": ["orders", "finance", "logistics"],
    },
    "customer": {
        "team": "Customer Platform Team",
        "source_systems": ["auth_service_postgres", "crm_salesforce", "billing_stripe"],
        "business_events": ["customer_registered", "subscription_started", "subscription_cancelled"],
        "analytical_questions_owned": [
            "What is our monthly active user count?",
            "What is the churn rate by cohort and acquisition channel?",
        ],
        "consumers": ["marketing", "finance", "support", "product"],
    },
}

# Anti-patterns to avoid:
# ✗ "Data" domain — too broad, recreates the central team problem
# ✗ "Reporting" domain — reporting is a consumer, not a producer
# ✗ System-aligned domains ("Postgres", "Kafka") — follow business capability, not technology
# ✗ Too granular (one domain per microservice) — domains should map to business sub-capabilities
#   not individual services. A domain may own multiple services.

Data Products — Designing the Interface Contract

A data product is not a pipeline, a table, or a dashboard. It is an independently deployable unit of data with an explicit owner, a versioned schema, documented SLOs, and discoverable metadata. The distinction matters: a pipeline is an implementation detail; a data product is a commitment to consumers. Consumers interact with the interface (schema, API, SLO), not with the internals (pipeline code, source system, transformation logic). This is exactly the contract model covered in detail in our article on Data Contracts and schema versioning.

The canonical data product attributes, from Dehghani's definition, are: discoverable, addressable, trustworthy, self-describing, interoperable, natively accessible, and value-generating. In practice, every data product needs four explicit artifacts: a schema definition (what columns, types, semantics), an SLO specification (freshness, completeness, accuracy targets and alert thresholds), an ownership manifest (team, on-call rotation, Slack channel), and a data catalog entry (description, lineage, usage examples). These four artifacts turn a table into a product.

# data-product.yaml — the manifest for a data product
# Checked into the domain team's repository, enforced by CI

apiVersion: datamesh.example.com/v1
kind: DataProduct
metadata:
  name: orders.order_events
  domain: orders
  owner: order-experience-squad
  oncall: "@orders-oncall"
  slack_channel: "#data-orders"
  catalog_url: "https://catalog.example.com/products/orders/order_events"

spec:
  description: |
    Immutable event log of all order lifecycle events: placement, cancellation,
    return initiation, and return resolution. One row per event. Append-only.
    Primary analytical source for order funnel metrics, cohort analysis, and
    refund rate reporting.

  interface:
    output_port: bigquery
    dataset: orders_prod
    table: order_events
    access_mode: read_via_authorized_view   # consumers get view, not raw table

  schema_version: "2.1.0"
  schema_registry: "https://schema-registry.example.com/subjects/orders.order_events"
  schema_evolution_policy: backward_compatible  # consumers are never broken silently

  slo:
    freshness:
      target_minutes: 30          # events appear within 30 minutes of occurrence
      warn_minutes: 45
      error_minutes: 90
    completeness:
      target_percent: 99.5        # >= 99.5% of events present within the freshness window
      warn_percent: 99.0
      error_percent: 97.0
    accuracy:
      validated_by: "dbt test orders.order_events"
      run_frequency: "every 1 hour"

  lineage:
    upstream_sources:
      - system: orders_service_postgres
        table: public.orders
        cdc_tool: debezium
      - system: orders_service_postgres
        table: public.returns
        cdc_tool: debezium
    downstream_consumers:
      - domain: finance
        product: finance.revenue_daily
      - domain: marketing
        product: marketing.campaign_attribution
      - domain: executive
        product: executive.kpi_dashboard

# Schema definition — versioned with backward compatibility guarantees

# orders/schemas/order_events_v2.avsc
{
  "type": "record",
  "name": "OrderEvent",
  "namespace": "com.example.orders",
  "doc": "An order lifecycle event. All monetary amounts are in minor units (cents).",
  "fields": [
    {"name": "event_id",       "type": "string",  "doc": "UUID v4, globally unique"},
    {"name": "event_type",     "type": {"type": "enum", "name": "EventType",
                                "symbols": ["ORDER_PLACED", "ORDER_CANCELLED",
                                            "RETURN_INITIATED", "RETURN_RESOLVED"]}},
    {"name": "event_timestamp","type": {"type": "long", "logicalType": "timestamp-micros"}},
    {"name": "order_id",       "type": "string"},
    {"name": "customer_id",    "type": "string"},
    {"name": "order_total_cents", "type": "long",  "doc": "Order total in minor currency units"},
    {"name": "currency_code",  "type": "string",   "default": "USD"},
    {"name": "channel",        "type": ["null", "string"], "default": null,
                               "doc": "Acquisition channel: web, ios, android, api"},
    {"name": "metadata",       "type": {"type": "map", "values": "string"}, "default": {},
                               "doc": "Extensible key-value pairs; do not use for queryable fields"}
  ]
}

# Schema evolution rules (enforced in CI via Confluent Schema Registry compatibility check):
# ALLOWED (backward compatible): add optional field with default, widen numeric type
# FORBIDDEN: remove field, rename field, change required field to different type
# REQUIRES MAJOR VERSION BUMP: rename event, restructure payload, remove enum value

Self-Serve Data Platform — Infrastructure as a Product

The self-serve platform is the enabling layer that makes domain ownership practical. Without it, distributing data ownership just distributes the complexity of running pipelines across dozens of teams that each reinvent the same Airflow setup, the same dbt profile configuration, the same Terraform module for BigQuery datasets. The platform team's job is to build and operate capabilities — not pipelines — that any domain team can use without asking for help.

The platform team's output is not dashboards or transformed tables; it is the tooling that domain teams use to build those things themselves. Concretely: a Terraform module that provisions a BigQuery dataset with the correct IAM bindings in one command, a dbt template repository that domain teams fork to get CI/CD, schema testing, and freshness checks pre-configured, and a data catalog with automatic registration on deployment. The platform team measures success by developer experience metrics — time to first data product, mean time to publish a schema change, number of support tickets from domain teams.

# Terraform module: provision a data product output port
# Used by domain teams — they fill in variables, platform team owns the module

# modules/data-product-output-port/main.tf

variable "domain"        { type = string }
variable "product_name"  { type = string }
variable "consumer_groups" {
  type    = list(string)
  default = []
  description = "List of IAM groups that get read access to the authorized view"
}

resource "google_bigquery_dataset" "product_dataset" {
  dataset_id  = "${var.domain}_${var.product_name}"
  location    = "EU"
  description = "Data product: ${var.domain}/${var.product_name}"

  labels = {
    domain      = var.domain
    product     = var.product_name
    managed_by  = "terraform"
    data_mesh   = "true"
  }

  # Domain team has full access to their own dataset
  access {
    role          = "OWNER"
    special_group = "projectOwners"
  }
}

# Consumers get access to authorized views only — not raw tables
resource "google_bigquery_dataset_iam_member" "consumer_access" {
  for_each   = toset(var.consumer_groups)
  dataset_id = google_bigquery_dataset.product_dataset.dataset_id
  role       = "roles/bigquery.dataViewer"
  member     = "group:${each.value}"
}

# Auto-register in the data catalog on apply
resource "null_resource" "catalog_registration" {
  triggers = {
    dataset_id = google_bigquery_dataset.product_dataset.dataset_id
  }

  provisioner "local-exec" {
    command = <<-EOT
      curl -X POST https://catalog.internal/api/v1/products         -H "Content-Type: application/json"         -d '{"domain": "${var.domain}", "product": "${var.product_name}",
             "output_port": "bigquery:${google_bigquery_dataset.product_dataset.id}"}'
    EOT
  }
}

# Domain team usage (no platform team involvement needed):
# module "order_events_product" {
#   source        = "git::https://github.com/example/platform//modules/data-product-output-port"
#   domain        = "orders"
#   product_name  = "order_events"
#   consumer_groups = ["finance-data@example.com", "marketing-data@example.com"]
# }

# dbt project template for domain teams
# Platform team maintains — domain teams inherit via template repository fork

# dbt_project.yml (domain team fills in name, version, models path)
name: 'orders_data_products'
version: '1.0.0'
config-version: 2

profile: 'orders'   # profile lives in ~/.dbt/profiles.yml, injected by CI

model-paths: ["models"]
test-paths:  ["tests"]
macro-paths: ["macros"]

# Platform-enforced defaults: all models must have tests and descriptions
models:
  orders_data_products:
    +materialized: table
    +on_schema_change: append_new_columns   # safe schema evolution default
    staging:
      +materialized: view
      +tags: ["staging"]
    products:
      +materialized: table
      +tags: ["data_product"]
      +meta:
        slo_freshness_minutes: 30
        slo_completeness_percent: 99.5
        owner: "order-experience-squad"

# Platform-provided macros (auto-available via packages.yml):
# {{ data_product_header() }}      — standard column header (inserted_at, source_system)
# {{ assert_freshness(column, minutes) }}  — reusable freshness test
# {{ assert_row_count_stable(table, pct_change) }} — row count regression test

Federated Computational Governance — Standards Without a Central Bottleneck

Federated governance solves the tension between domain autonomy and organizational standards. Domain teams own their data products, but the organization needs consistent data classification, interoperable schemas, shared vocabulary for common entities (customer, order, product), privacy compliance, and access control that works across domain boundaries. The governance model must enforce standards without becoming a gatekeeper that recreates the central bottleneck.

The practical mechanism is policy-as-code: governance rules are expressed as automated checks that run in every domain team's CI pipeline, not as approval gates requiring a human reviewer. The governance council sets the standards; the platform team encodes them as code; domain teams' deployments fail if standards are violated. For access control, tools like Databricks Unity Catalog provide column-level masking policies and row filters that the governance team configures centrally while domain teams operate independently within those guardrails.

# policy-as-code: governance checks that run in every domain team's CI

# .github/workflows/data-product-governance.yml
name: Data Product Governance Checks
on: [pull_request]

jobs:
  governance:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Validate data product manifest
        run: |
          python scripts/validate_manifest.py data-product.yaml

      - name: Check schema backward compatibility
        run: |
          # Pull current schema from registry, check compatibility of proposed schema
          docker run --rm             -e SCHEMA_REGISTRY_URL=${SCHEMA_REGISTRY_URL}             -v $(pwd)/schemas:/schemas             confluentinc/cp-schema-registry:latest             kafka-schema-registry-run-class io.confluent.kafka.schemaregistry.tools.SchemaRegistryPerformance             --check-compatibility /schemas/order_events.avsc

      - name: Enforce PII column tagging
        run: |
          # All columns matching PII patterns must have classification tags
          python scripts/pii_tag_check.py models/ schemas/

      - name: Validate SLO thresholds
        run: |
          # Freshness SLO must be defined and within policy bounds
          python scripts/slo_check.py data-product.yaml --max-freshness-minutes 120

      - name: Check consumer access declarations
        run: |
          # Consumers must be declared in the manifest before access is granted
          python scripts/consumer_access_check.py data-product.yaml

# scripts/validate_manifest.py — policy-as-code for data product manifests

import sys
import yaml
from dataclasses import dataclass
from typing import Optional

REQUIRED_FIELDS = [
    "metadata.name", "metadata.domain", "metadata.owner",
    "metadata.oncall", "spec.slo.freshness.target_minutes",
    "spec.slo.completeness.target_percent", "spec.lineage.upstream_sources",
]

PII_COLUMN_PATTERNS = [
    "email", "phone", "name", "address", "ip_address",
    "date_of_birth", "ssn", "passport", "national_id",
]

def get_nested(d: dict, path: str):
    keys = path.split(".")
    current = d
    for key in keys:
        if not isinstance(current, dict) or key not in current:
            return None
        current = current[key]
    return current

def validate(manifest_path: str) -> list[str]:
    with open(manifest_path) as f:
        manifest = yaml.safe_load(f)

    errors = []

    # Check required fields
    for field in REQUIRED_FIELDS:
        if get_nested(manifest, field) is None:
            errors.append(f"Missing required field: {field}")

    # Check freshness SLO is within policy bounds
    freshness = get_nested(manifest, "spec.slo.freshness.target_minutes")
    if freshness and freshness > 120:
        errors.append(
            f"Freshness SLO target {freshness}m exceeds policy maximum of 120m. "
            "Request an exemption via the governance board."
        )

    # Check schema_version follows semver
    schema_version = get_nested(manifest, "spec.schema_version")
    if schema_version:
        parts = str(schema_version).split(".")
        if len(parts) != 3 or not all(p.isdigit() for p in parts):
            errors.append(f"schema_version '{schema_version}' must follow semantic versioning (MAJOR.MINOR.PATCH)")

    # Check that PII columns are declared in the sensitivity section
    # (Real implementation would cross-reference against the schema registry)
    sensitivity = get_nested(manifest, "spec.sensitivity") or {}
    pii_columns = sensitivity.get("pii_columns", [])
    if not pii_columns and get_nested(manifest, "metadata.domain") == "customer":
        errors.append(
            "Customer domain products must declare pii_columns in spec.sensitivity. "
            "See https://governance.internal/pii-policy for the column inventory."
        )

    return errors

if __name__ == "__main__":
    manifest_path = sys.argv[1]
    errors = validate(manifest_path)
    if errors:
        print("\nGovernance validation FAILED:")
        for err in errors:
            print(f"  ✗ {err}")
        sys.exit(1)
    print("Governance validation passed.")

Implementing a Data Product with dbt — End to End

A complete data product implementation in dbt consists of three layers: a staging layer that mirrors the source system schema with minimal transformation, an intermediate layer that applies business logic and joins, and a product layer that exposes the contract-compliant interface to consumers. The product layer is the only layer consumers access; the staging and intermediate layers are internal implementation details that the domain team can refactor freely.

# models/staging/stg_orders__orders.sql
-- Staging: mirror source schema, add standard metadata columns, no business logic

{{
  config(
    materialized = 'view',
    tags = ['staging', 'orders']
  )
}}

with source as (
    -- Reads from CDC-replicated raw table (Debezium → BigQuery via Datastream)
    select * from {{ source('orders_cdc', 'orders') }}
    where _cdc_deleted = false      -- exclude soft-deleted rows
),

renamed as (
    select
        id                      as order_id,
        customer_id,
        status,
        total_amount_cents,
        currency_code,
        channel,
        created_at              as placed_at,
        updated_at,
        _cdc_source_timestamp   as source_updated_at,
        current_timestamp()     as dbt_inserted_at,
        '{{ invocation_id }}'   as dbt_run_id

    from source
)

select * from renamed

# models/products/order_events.sql
-- Product layer: the contracted interface exposed to consumers
-- Schema must match order_events.avsc exactly

{{
  config(
    materialized = 'table',
    partition_by = {
      "field": "event_date",
      "data_type": "date",
      "granularity": "day"
    },
    cluster_by = ["customer_id", "event_type"],
    tags = ['data_product', 'orders'],
    meta = {
      "product_name": "orders.order_events",
      "schema_version": "2.1.0",
      "slo_freshness_minutes": 30,
      "owner": "order-experience-squad"
    }
  )
}}

with orders as (
    select * from {{ ref('stg_orders__orders') }}
),

returns as (
    select * from {{ ref('stg_orders__returns') }}
),

-- Combine order placements and return events into a unified event stream
order_placed_events as (
    select
        {{ dbt_utils.generate_surrogate_key(['order_id', "'ORDER_PLACED'"]) }}
                                    as event_id,
        'ORDER_PLACED'              as event_type,
        placed_at                   as event_timestamp,
        date(placed_at)             as event_date,
        order_id,
        customer_id,
        total_amount_cents          as order_total_cents,
        currency_code,
        channel,
        struct(
          'placed_at' as key,
          cast(placed_at as string) as value
        )                           as metadata
    from orders
    where status != 'TEST'
),

order_cancelled_events as (
    select
        {{ dbt_utils.generate_surrogate_key(['order_id', "'ORDER_CANCELLED'"]) }}
                                    as event_id,
        'ORDER_CANCELLED'           as event_type,
        updated_at                  as event_timestamp,
        date(updated_at)            as event_date,
        order_id,
        customer_id,
        total_amount_cents          as order_total_cents,
        currency_code,
        channel,
        struct('status' as key, status as value) as metadata
    from orders
    where status = 'CANCELLED'
),

return_events as (
    select
        {{ dbt_utils.generate_surrogate_key(['return_id', "event_type"]) }}
                                    as event_id,
        event_type,
        event_timestamp,
        date(event_timestamp)       as event_date,
        order_id,
        customer_id,
        refund_amount_cents         as order_total_cents,
        currency_code,
        null                        as channel,
        struct('return_id' as key, return_id as value) as metadata
    from {{ ref('stg_orders__returns') }}
    where event_type in ('RETURN_INITIATED', 'RETURN_RESOLVED')
),

unioned as (
    select * from order_placed_events
    union all
    select * from order_cancelled_events
    union all
    select * from return_events
)

select * from unioned

# models/products/order_events.yml
-- Declarative tests that enforce the data product SLO

version: 2

models:
  - name: order_events
    description: |
      Immutable event log of all order lifecycle events. One row per event.
      Append-only. Schema version 2.1.0. See data-product.yaml for full SLO.
    columns:
      - name: event_id
        description: UUID v4, globally unique across all event types
        data_tests:
          - unique
          - not_null
      - name: event_type
        data_tests:
          - not_null
          - accepted_values:
              values: ['ORDER_PLACED', 'ORDER_CANCELLED', 'RETURN_INITIATED', 'RETURN_RESOLVED']
      - name: event_timestamp
        data_tests:
          - not_null
          - dbt_utils.recency:
              datepart: minute
              field: event_timestamp
              interval: 45    # SLO freshness warn threshold
              severity: warn
          - dbt_utils.recency:
              datepart: minute
              field: event_timestamp
              interval: 90    # SLO freshness error threshold
              severity: error
      - name: order_id
        data_tests:
          - not_null
      - name: customer_id
        data_tests:
          - not_null
      - name: order_total_cents
        data_tests:
          - not_null
          - dbt_utils.expression_is_true:
              expression: ">= 0"

    data_tests:
      # Row count regression — new run should not drop more than 0.5% of rows
      - dbt_utils.equal_rowcount:
          compare_model: ref('order_events')  # compare to previous materialization
          severity: warn

      # Referential integrity: every order_id must exist in orders staging
      - dbt_utils.relationships_where:
          to: ref('stg_orders__orders')
          field: order_id
          from_condition: "event_type = 'ORDER_PLACED'"
          to_condition: "status != 'TEST'"

Measuring Data Mesh Maturity — DORA Metrics for Data Products

Data Mesh adoption should be measured empirically. The same DORA metrics that software engineering teams use for deployment health can be adapted to data products: deployment frequency (how often does a domain publish data product updates?), lead time for changes (how long from a source system change to a consumer-visible update?), change failure rate (what fraction of data product deployments cause downstream breakages?), and mean time to restore (how long to recover from a data product SLO violation?).

# data_product_metrics.py
# Emitted to your observability platform from CI/CD and monitoring

import time
from opentelemetry import metrics
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import OTLPMetricExporter

meter_provider = MeterProvider(
    metric_readers=[
        PeriodicExportingMetricReader(OTLPMetricExporter(), export_interval_millis=60_000)
    ]
)
meter = meter_provider.get_meter("datamesh.metrics")

# DORA-inspired data product metrics
deployment_counter = meter.create_counter(
    name="datamesh.product.deployments",
    description="Number of data product deployments",
    unit="1",
)

lead_time_histogram = meter.create_histogram(
    name="datamesh.product.lead_time_minutes",
    description="Lead time from source change to consumer-visible update",
    unit="min",
)

slo_violation_counter = meter.create_counter(
    name="datamesh.product.slo_violations",
    description="Number of SLO violations (freshness, completeness, accuracy)",
    unit="1",
)

mttr_histogram = meter.create_histogram(
    name="datamesh.product.mttr_minutes",
    description="Mean time to restore from SLO violation",
    unit="min",
)

consumer_satisfaction_gauge = meter.create_observable_gauge(
    name="datamesh.product.consumer_satisfaction_score",
    description="Consumer satisfaction score from quarterly survey (0-10 NPS)",
    callbacks=[lambda obs: obs.observe(8.2, {"domain": "orders", "product": "order_events"})],
)

# Emitted on each dbt run completion (from dbt run_results.json processing)
def record_deployment(domain: str, product: str, lead_time_minutes: float, success: bool):
    attrs = {"domain": domain, "product": product}
    deployment_counter.add(1, {**attrs, "status": "success" if success else "failure"})
    if success:
        lead_time_histogram.record(lead_time_minutes, attrs)

# Emitted by the SLO monitoring job (runs every 5 minutes)
def record_slo_violation(domain: str, product: str, slo_type: str, severity: str):
    slo_violation_counter.add(1, {
        "domain": domain, "product": product,
        "slo_type": slo_type, "severity": severity
    })

# Emitted when an incident is resolved
def record_incident_resolved(domain: str, product: str, mttr_minutes: float):
    mttr_histogram.record(mttr_minutes, {"domain": domain, "product": product})


# Grafana dashboard alerts (PromQL / MQL):
#
# Freshness SLO breach rate > 1% of the time in the last 7 days:
# rate(datamesh_product_slo_violations_total{slo_type="freshness"}[7d])
#   / rate(datamesh_product_deployments_total[7d]) > 0.01
#
# Lead time regression — p95 lead time increased by > 20% vs previous week:
# histogram_quantile(0.95, datamesh_product_lead_time_minutes_bucket[1d])
#   / histogram_quantile(0.95, datamesh_product_lead_time_minutes_bucket[1d] offset 7d) > 1.2