diff --git a/README.md b/README.md index 7b42217..5a18401 100644 --- a/README.md +++ b/README.md @@ -519,21 +519,21 @@ The repository currently includes: - a completed Phase 8.1 asset/model/semantic/metric lineage graph with transitive impact analysis; - a completed Phase 8.2 field-level lineage graph from DFF source fields through dbt derivations to governed metrics; - a completed Phase 8.3 trust overlay that projects dbt checks, quality-contract rules, and durable incidents onto exact lineage nodes; -- a Phase 8.4 runtime-lineage workbench that links the latest Airflow DAG/task execution to its replay-safe Studio pipeline run, structured steps, source asset, and curated target; +- a completed Phase 8.4 runtime-lineage workbench that links the latest Airflow DAG/task execution to its replay-safe Studio pipeline run, structured steps, source asset, and curated target; +- a Phase 8.5 analytics-consumer lineage surface that resolves source-controlled report dependencies against governed dbt semantic metrics without claiming downstream deployment; - backend/frontend validation, repository map, and compact handoff tooling. -The active implementation target is **Phase 8.4 — Pipeline and Airflow Execution Lineage Foundation**. Phases 0 through 7 are complete; Phase 8.1 established asset-level impact, Phase 8.2 added field-level impact, and Phase 8.3 projected trust evidence onto those nodes. Phase 8.4 now joins the latest Airflow DAG/task proof to the replay-safe Studio pipeline run and its structured materialization steps without reading Airflow metadata tables directly. +The active implementation target is **Phase 8.5 — Analytics Consumer Lineage and Impact Closure**. Phases 0 through 7 are complete; Phase 8.1 established asset impact, Phase 8.2 added field impact, Phase 8.3 projected trust evidence, and Phase 8.4 joined Airflow/Studio runtime execution to the structural graph. Phase 8.5 now closes the lineage chain at source-controlled analytics consumers while keeping actual analytical delivery and Power BI deployment in their later roadmap phases. -## Phase 8.4 quick start +## Phase 8.5 quick start -Phase 8.4 federates runtime evidence into lineage without moving execution ownership. Airflow remains authoritative for DAG-run and task-instance state; Studio remains authoritative for pipeline-run and step evidence. The runtime graph links the existing `DFF` and `FED_FUNDS_RATE_MART` structural nodes to the latest `AIRFLOW:` Studio run. +Phase 8.5 resolves declared analytics dependencies against dbt-owned semantic artifacts. The first proof declares a Federal Funds Rate Overview report that requires all four governed Federal Funds Rate metrics. The report targets Power BI but remains `DECLARED`; no Power BI deployment is claimed. ```powershell -# Ensure the Airflow proof runtime is available, then start the Studio API/web as usual. Invoke-RestMethod ` - "http://localhost:8100/api/v1/lineage/runtime/summary" | + "http://localhost:8100/api/v1/lineage/consumers/summary" | ConvertTo-Json -Depth 12 ``` -A fully connected proof reports `runtime_status=READY`, four successful Airflow tasks, four successful Studio steps, and replay-safe materialization evidence for `mart.fed_funds_rate`. +A complete proof reports one resolved consumer, four resolved metrics, zero unresolved dependencies, 5 nodes, and 4 `CONSUMED_BY` edges. diff --git a/apps/api/skydata_studio/api/routes/lineage.py b/apps/api/skydata_studio/api/routes/lineage.py index 96d52d0..1058faf 100644 --- a/apps/api/skydata_studio/api/routes/lineage.py +++ b/apps/api/skydata_studio/api/routes/lineage.py @@ -6,6 +6,8 @@ from skydata_studio.core.config import Settings, get_settings from skydata_studio.db.session import SessionDependency from skydata_studio.schemas.lineage import ( + AnalyticsConsumerImpactSummary, + AnalyticsConsumerLineageSummary, FieldLineageImpactSummary, FieldLineageSummary, LineageImpactSummary, @@ -19,6 +21,10 @@ lineage_impact, lineage_summary, ) +from skydata_studio.services.lineage_consumers import ( + analytics_consumer_impact, + analytics_consumer_lineage_summary, +) from skydata_studio.services.lineage_runtime import runtime_lineage_summary from skydata_studio.services.lineage_trust import lineage_trust_summary @@ -96,3 +102,23 @@ def lineage_runtime_execution( return runtime_lineage_summary(session, settings) except (SQLAlchemyError, OSError, ValueError, TypeError) as error: raise _unavailable(error) from error + + +@router.get("/consumers/summary", response_model=AnalyticsConsumerLineageSummary) +def lineage_analytics_consumers( + focus_metric_name: Annotated[str | None, Query(alias="focusMetricName")] = None, +) -> AnalyticsConsumerLineageSummary: + try: + return analytics_consumer_lineage_summary(focus_metric_name=focus_metric_name) + except (OSError, ValueError, TypeError) as error: + raise _unavailable(error) from error + + +@router.get("/consumers/impact", response_model=AnalyticsConsumerImpactSummary) +def lineage_analytics_consumer_impact( + metric_name: Annotated[str, Query(alias="metricName", min_length=1)], +) -> AnalyticsConsumerImpactSummary: + try: + return analytics_consumer_impact(metric_name) + except (OSError, ValueError, TypeError) as error: + raise _unavailable(error) from error diff --git a/apps/api/skydata_studio/api/routes/platform.py b/apps/api/skydata_studio/api/routes/platform.py index 942f10b..8812317 100644 --- a/apps/api/skydata_studio/api/routes/platform.py +++ b/apps/api/skydata_studio/api/routes/platform.py @@ -96,7 +96,7 @@ def platform_summary() -> PlatformSummary: product="SkyData Studio", subtitle="Data Engineering Workbench", theme="Aurora Foundry", - current_phase="Phase 8.4 — Pipeline and Airflow Execution Lineage Foundation", + current_phase="Phase 8.5 — Analytics Consumer Lineage and Impact Closure", boundary=( "SkyData Studio starts after SkyCommand ingestion and publishes governed " "analytical products for SkyWeb Analytics, Power BI, and future consumers." diff --git a/apps/api/skydata_studio/schemas/lineage.py b/apps/api/skydata_studio/schemas/lineage.py index e200a1b..2102f81 100644 --- a/apps/api/skydata_studio/schemas/lineage.py +++ b/apps/api/skydata_studio/schemas/lineage.py @@ -241,3 +241,74 @@ class RuntimeLineageSummary(BaseModel): nodes: list[RuntimeLineageNode] = Field(default_factory=list) edges: list[RuntimeLineageEdge] = Field(default_factory=list) + +AnalyticsConsumerType = Literal["REPORT", "APPLICATION"] +AnalyticsConsumerDeploymentStatus = Literal["DECLARED", "DEPLOYED", "RETIRED"] +AnalyticsConsumerResolutionStatus = Literal["READY", "PARTIAL", "MISSING"] + + +class AnalyticsConsumerDefinition(BaseModel): + code: str + version: str + name: str + description: str + consumer_type: AnalyticsConsumerType + delivery_system: str + deployment_status: AnalyticsConsumerDeploymentStatus + semantic_model: str + required_metrics: list[str] + required_dimensions: list[str] = Field(default_factory=list) + owner: str + + +class AnalyticsConsumerMetricBinding(BaseModel): + consumer_code: str + metric_name: str + metric_label: str | None = None + metric_unique_id: str | None = None + resolved: bool + message: str + + +class AnalyticsConsumerNode(BaseModel): + id: str + label: str + node_type: Literal["METRIC", "ANALYTICS_CONSUMER"] + system: str + status: str + metadata: dict[str, object] = Field(default_factory=dict) + + +class AnalyticsConsumerEdge(BaseModel): + id: str + upstream_id: str + downstream_id: str + edge_type: Literal["CONSUMED_BY"] + label: str + + +class AnalyticsConsumerImpactSummary(BaseModel): + selected_metric_name: str | None = None + selected_metric_label: str | None = None + downstream_consumer_count: int = Field(default=0, ge=0) + consumers: list[AnalyticsConsumerNode] = Field(default_factory=list) + + +class AnalyticsConsumerLineageSummary(BaseModel): + project_name: str = "skydata_studio" + phase: str = "Phase 8.5 — Analytics Consumer Lineage and Impact Closure" + consumer_status: AnalyticsConsumerResolutionStatus + semantic_artifact_status: Literal["READY", "PENDING", "MISSING"] + consumer_contract_count: int = Field(default=0, ge=0) + resolved_consumer_count: int = Field(default=0, ge=0) + declared_metric_count: int = Field(default=0, ge=0) + resolved_metric_count: int = Field(default=0, ge=0) + unresolved_metric_count: int = Field(default=0, ge=0) + node_count: int = Field(default=0, ge=0) + edge_count: int = Field(default=0, ge=0) + source_paths: list[str] = Field(default_factory=list) + consumers: list[AnalyticsConsumerDefinition] = Field(default_factory=list) + metric_bindings: list[AnalyticsConsumerMetricBinding] = Field(default_factory=list) + nodes: list[AnalyticsConsumerNode] = Field(default_factory=list) + edges: list[AnalyticsConsumerEdge] = Field(default_factory=list) + default_impact: AnalyticsConsumerImpactSummary diff --git a/apps/api/skydata_studio/services/lineage_consumers.py b/apps/api/skydata_studio/services/lineage_consumers.py new file mode 100644 index 0000000..c412c10 --- /dev/null +++ b/apps/api/skydata_studio/services/lineage_consumers.py @@ -0,0 +1,262 @@ +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any, Literal, cast + +from skydata_studio.schemas.dbt import DbtSemanticLayerSummary +from skydata_studio.schemas.lineage import ( + AnalyticsConsumerDefinition, + AnalyticsConsumerEdge, + AnalyticsConsumerImpactSummary, + AnalyticsConsumerLineageSummary, + AnalyticsConsumerMetricBinding, + AnalyticsConsumerNode, +) +from skydata_studio.services.dbt_transformations import dbt_semantic_layer + + +def _repository_root() -> Path: + return Path(__file__).resolve().parents[4] + + +def _contract_directory() -> Path: + return _repository_root() / "contracts" / "analytics" + + +def _read_consumer(path: Path) -> AnalyticsConsumerDefinition: + with path.open(encoding="utf-8") as handle: + payload = json.load(handle) + if not isinstance(payload, dict): + raise ValueError(f"Expected an object in analytics consumer contract {path}.") + return AnalyticsConsumerDefinition.model_validate(cast(dict[str, Any], payload)) + + +def _read_consumers( + directory: Path | None = None, +) -> tuple[list[AnalyticsConsumerDefinition], list[str]]: + root = directory or _contract_directory() + if not root.exists(): + return [], [] + paths = sorted(root.glob("*.json")) + consumers = [_read_consumer(path) for path in paths] + source_paths: list[str] = [] + for path in paths: + try: + source_paths.append(path.relative_to(_repository_root()).as_posix()) + except ValueError: + source_paths.append(path.as_posix()) + return consumers, source_paths + + +def _consumer_node_id(code: str) -> str: + return f"consumer:{code.lower()}" + + +def _metric_node_id(unique_id: str) -> str: + return f"metric:{unique_id}" + + +def _impact_from_graph( + *, + metric_name: str | None, + nodes: list[AnalyticsConsumerNode], + edges: list[AnalyticsConsumerEdge], +) -> AnalyticsConsumerImpactSummary: + metric_nodes = { + str(node.metadata.get("metric_name") or ""): node + for node in nodes + if node.node_type == "METRIC" + } + selected_name = metric_name if metric_name in metric_nodes else None + if selected_name is None and metric_nodes: + selected_name = sorted(metric_nodes)[0] + if selected_name is None: + return AnalyticsConsumerImpactSummary() + + selected = metric_nodes[selected_name] + downstream_ids = { + edge.downstream_id for edge in edges if edge.upstream_id == selected.id + } + consumers = [ + node + for node in nodes + if node.node_type == "ANALYTICS_CONSUMER" and node.id in downstream_ids + ] + consumers.sort(key=lambda item: item.label) + return AnalyticsConsumerImpactSummary( + selected_metric_name=selected_name, + selected_metric_label=selected.label, + downstream_consumer_count=len(consumers), + consumers=consumers, + ) + + +def compose_consumer_lineage_summary( + *, + consumers: list[AnalyticsConsumerDefinition], + semantic: DbtSemanticLayerSummary, + source_paths: list[str] | None = None, + focus_metric_name: str | None = None, +) -> AnalyticsConsumerLineageSummary: + metric_by_name = {metric.name: metric for metric in semantic.metrics} + semantic_model_by_name = {model.name: model for model in semantic.semantic_models} + nodes_by_id: dict[str, AnalyticsConsumerNode] = {} + edges: list[AnalyticsConsumerEdge] = [] + bindings: list[AnalyticsConsumerMetricBinding] = [] + resolved_consumers = 0 + declared_metric_count = 0 + resolved_metric_count = 0 + + for consumer in consumers: + declared_metric_count += len(consumer.required_metrics) + semantic_model = semantic_model_by_name.get(consumer.semantic_model) + available_dimensions = { + dimension.name for dimension in semantic_model.dimensions + } if semantic_model is not None else set() + dimensions_ready = all( + dimension in available_dimensions for dimension in consumer.required_dimensions + ) + consumer_metrics_ready = True + consumer_node_id = _consumer_node_id(consumer.code) + nodes_by_id[consumer_node_id] = AnalyticsConsumerNode( + id=consumer_node_id, + label=consumer.name, + node_type="ANALYTICS_CONSUMER", + system=consumer.delivery_system, + status="DECLARED", + metadata={ + "consumer_code": consumer.code, + "consumer_type": consumer.consumer_type, + "deployment_status": consumer.deployment_status, + "semantic_model": consumer.semantic_model, + "owner": consumer.owner, + "required_dimensions": len(consumer.required_dimensions), + }, + ) + + for metric_name in consumer.required_metrics: + metric = metric_by_name.get(metric_name) + resolved = ( + semantic.artifact_status == "READY" + and semantic_model is not None + and metric is not None + and metric.semantic_model == consumer.semantic_model + ) + bindings.append( + AnalyticsConsumerMetricBinding( + consumer_code=consumer.code, + metric_name=metric_name, + metric_label=metric.label if metric is not None else None, + metric_unique_id=metric.unique_id if metric is not None else None, + resolved=resolved, + message=( + "Governed metric resolves to the declared semantic consumer." + if resolved + else "Declared metric is not available on the required semantic model." + ), + ) + ) + if not resolved or metric is None: + consumer_metrics_ready = False + continue + + resolved_metric_count += 1 + metric_node_id = _metric_node_id(metric.unique_id) + nodes_by_id.setdefault( + metric_node_id, + AnalyticsConsumerNode( + id=metric_node_id, + label=metric.label, + node_type="METRIC", + system="dbt semantic", + status="READY", + metadata={ + "metric_name": metric.name, + "metric_type": metric.metric_type, + "semantic_model": metric.semantic_model, + }, + ), + ) + edges.append( + AnalyticsConsumerEdge( + id=f"consumer-metric:{metric_node_id}:{consumer_node_id}", + upstream_id=metric_node_id, + downstream_id=consumer_node_id, + edge_type="CONSUMED_BY", + label="declared report dependency", + ) + ) + + if ( + semantic.artifact_status == "READY" + and semantic_model is not None + and dimensions_ready + and consumer_metrics_ready + ): + resolved_consumers += 1 + nodes_by_id[consumer_node_id].status = "READY" + + unresolved_metric_count = declared_metric_count - resolved_metric_count + if consumers and resolved_consumers == len(consumers) and not unresolved_metric_count: + consumer_status: Literal["READY", "PARTIAL", "MISSING"] = "READY" + elif consumers or nodes_by_id: + consumer_status = "PARTIAL" + else: + consumer_status = "MISSING" + + nodes = sorted( + nodes_by_id.values(), + key=lambda item: (item.node_type == "ANALYTICS_CONSUMER", item.label), + ) + default_impact = _impact_from_graph( + metric_name=focus_metric_name, + nodes=nodes, + edges=edges, + ) + return AnalyticsConsumerLineageSummary( + consumer_status=consumer_status, + semantic_artifact_status=semantic.artifact_status, + consumer_contract_count=len(consumers), + resolved_consumer_count=resolved_consumers, + declared_metric_count=declared_metric_count, + resolved_metric_count=resolved_metric_count, + unresolved_metric_count=unresolved_metric_count, + node_count=len(nodes), + edge_count=len(edges), + source_paths=source_paths or [], + consumers=consumers, + metric_bindings=bindings, + nodes=nodes, + edges=edges, + default_impact=default_impact, + ) + + +def analytics_consumer_lineage_summary( + *, + focus_metric_name: str | None = None, + contract_directory: Path | None = None, + target_dir: Path | None = None, +) -> AnalyticsConsumerLineageSummary: + consumers, source_paths = _read_consumers(contract_directory) + return compose_consumer_lineage_summary( + consumers=consumers, + semantic=dbt_semantic_layer(target_dir), + source_paths=source_paths, + focus_metric_name=focus_metric_name, + ) + + +def analytics_consumer_impact( + metric_name: str, + *, + contract_directory: Path | None = None, + target_dir: Path | None = None, +) -> AnalyticsConsumerImpactSummary: + summary = analytics_consumer_lineage_summary( + focus_metric_name=metric_name, + contract_directory=contract_directory, + target_dir=target_dir, + ) + return summary.default_impact diff --git a/apps/web/src/App.jsx b/apps/web/src/App.jsx index 5ab545d..41ed63c 100644 --- a/apps/web/src/App.jsx +++ b/apps/web/src/App.jsx @@ -35,7 +35,7 @@ const PAGE_META = { '/quality/contracts': ['Contracts', 'Quality gates and consumer compatibility'], '/quality/incidents': ['Quality Incidents', 'Durable failures, ownership, and remediation history'], '/quality/reliability': ['Reliability', 'Quality SLO, observation history, and consumer-readiness posture'], - '/quality/lineage': ['Lineage', 'Asset, field, trust, and change impact'], + '/quality/lineage': ['Lineage', 'Asset, field, trust, runtime, and consumer impact'], '/delivery/marts': ['Analytical Marts', 'Curated consumer-ready facts and dimensions'], '/delivery/semantic': ['Semantic Layer', 'Governed metrics and business definitions'], '/delivery/reports': ['Reports', 'Reporting inventory and refresh evidence'], diff --git a/apps/web/src/components/ConsumerLineagePanel.jsx b/apps/web/src/components/ConsumerLineagePanel.jsx new file mode 100644 index 0000000..d60b0f9 --- /dev/null +++ b/apps/web/src/components/ConsumerLineagePanel.jsx @@ -0,0 +1,180 @@ +import { useCallback, useEffect, useMemo, useState } from 'react'; +import StatusPill from './StatusPill.jsx'; +import { getJson } from '../services/api.js'; + +const EMPTY = { + consumer_status: 'MISSING', + semantic_artifact_status: 'MISSING', + consumer_contract_count: 0, + resolved_consumer_count: 0, + declared_metric_count: 0, + resolved_metric_count: 0, + unresolved_metric_count: 0, + node_count: 0, + edge_count: 0, + source_paths: [], + consumers: [], + metric_bindings: [], + nodes: [], + edges: [], + default_impact: { + selected_metric_name: null, + selected_metric_label: null, + downstream_consumer_count: 0, + consumers: [], + }, +}; + +function consumerTone(status) { + if (status === 'READY') return 'READY'; + if (status === 'PARTIAL') return 'WARNING'; + return 'PLANNED'; +} + +function ConsumerLineagePanel() { + const [summary, setSummary] = useState(EMPTY); + const [impact, setImpact] = useState(EMPTY.default_impact); + const [loading, setLoading] = useState(true); + const [impactLoading, setImpactLoading] = useState(false); + const [error, setError] = useState(''); + + const load = useCallback(async () => { + setLoading(true); + setError(''); + try { + const payload = await getJson('/api/v1/lineage/consumers/summary'); + setSummary(payload); + setImpact(payload.default_impact || EMPTY.default_impact); + } catch (requestError) { + setError(requestError.message); + } finally { + setLoading(false); + } + }, []); + + useEffect(() => { + const controller = new AbortController(); + getJson('/api/v1/lineage/consumers/summary', { signal: controller.signal }) + .then((payload) => { + setSummary(payload); + setImpact(payload.default_impact || EMPTY.default_impact); + }) + .catch((requestError) => { + if (requestError.name !== 'AbortError') setError(requestError.message); + }) + .finally(() => { + if (!controller.signal.aborted) setLoading(false); + }); + return () => controller.abort(); + }, []); + + const metricNodes = useMemo( + () => summary.nodes.filter((node) => node.node_type === 'METRIC'), + [summary.nodes], + ); + const consumerNodes = useMemo( + () => summary.nodes.filter((node) => node.node_type === 'ANALYTICS_CONSUMER'), + [summary.nodes], + ); + + async function selectMetric(node) { + const metricName = node.metadata.metric_name; + if (!metricName) return; + setImpactLoading(true); + setError(''); + try { + setImpact(await getJson( + `/api/v1/lineage/consumers/impact?metricName=${encodeURIComponent(metricName)}`, + )); + } catch (requestError) { + setError(requestError.message); + } finally { + setImpactLoading(false); + } + } + + return ( +
+
+
+ PHASE 8.5 · ANALYTICS CONSUMER LINEAGE +

Governed metrics terminate at declared consumers

+

+ Resolve source-controlled report dependencies against dbt semantic artifacts without + claiming a Power BI deployment that Phase 10 has not provisioned yet. +

+
+
+ + +
+
+ + {error ?
!{error}
: null} + +
+
CONSUMERS{summary.resolved_consumer_count}/{summary.consumer_contract_count}source-controlled declarations
+
METRICS{summary.resolved_metric_count}/{summary.declared_metric_count}{summary.unresolved_metric_count} unresolved
+
SEMANTIC{summary.semantic_artifact_status}dbt-owned evidence
+
GRAPH{summary.node_count}{summary.edge_count} consumer edges
+
+ +
+
+
GOVERNED METRICSSelect a metric to inspect consumers
+
+ {metricNodes.map((node) => ( + + ))} +
+
+ +
+
DECLARED CONSUMERSReporting dependency contracts
+
+ {consumerNodes.map((node) => ( +
+ {node.metadata.consumer_type || 'CONSUMER'} + {node.label} + {node.system} · {node.metadata.deployment_status || node.status} +
+ ))} +
+
+
+ +
+
+ SELECTED METRIC + {impact.selected_metric_label || 'No metric selected'} +
+
+ DOWNSTREAM CONSUMERS + {impactLoading ? '…' : impact.downstream_consumer_count} +
+
+ DELIVERY BOUNDARY + DECLARATION ONLY +
+
+ +
+ {summary.edges.map((edge) => ( + {edge.edge_type}{edge.label} + ))} +
+
+ ); +} + +export default ConsumerLineagePanel; diff --git a/apps/web/src/components/Sidebar.jsx b/apps/web/src/components/Sidebar.jsx index a1c714a..f3f5f45 100644 --- a/apps/web/src/components/Sidebar.jsx +++ b/apps/web/src/components/Sidebar.jsx @@ -122,7 +122,7 @@ function Sidebar({ mobileOpen, onClose }) { Development - Phase 8.4 execution lineage + Phase 8.5 consumer lineage diff --git a/apps/web/src/pages/Lineage.jsx b/apps/web/src/pages/Lineage.jsx index fa37258..bb4197e 100644 --- a/apps/web/src/pages/Lineage.jsx +++ b/apps/web/src/pages/Lineage.jsx @@ -2,6 +2,7 @@ import { useCallback, useEffect, useMemo, useState } from 'react'; import FieldLineagePanel from '../components/FieldLineagePanel.jsx'; import LineageTrustPanel from '../components/LineageTrustPanel.jsx'; import RuntimeLineagePanel from '../components/RuntimeLineagePanel.jsx'; +import ConsumerLineagePanel from '../components/ConsumerLineagePanel.jsx'; import StatusPill from '../components/StatusPill.jsx'; import { getJson } from '../services/api.js'; @@ -121,11 +122,11 @@ function Lineage() {
- PHASE 8.4 · PIPELINE + AIRFLOW EXECUTION LINEAGE + PHASE 8.5 · ANALYTICS CONSUMER LINEAGE + IMPACT CLOSURE

Lineage & Impact

- Preserve asset, field, and trust evidence from Phases 8.1–8.3, then connect the latest Airflow - and Studio execution proof to the same source and curated target nodes. + Preserve structural, field, trust, and runtime evidence from Phases 8.1–8.4, then terminate the + graph at source-controlled analytics consumers that declare exactly which governed metrics they use.