Echo Service¶
The Echo service is the unified telemetry hub — it receives and exposes pipeline state, metrics, lineage, events, quality results, and recovery metadata.
Multi-Channel Ingestion¶
Echo accepts state updates through three channels:
graph LR
REST[REST API] --> ECHO[Echo Service]
WH[Inbound Webhooks] --> ECHO
EB[Event Bus] --> ECHO
ECHO --> STATE[State Store]
ECHO --> NOTIFY[Webhook Notifications]| Channel | Use Case |
|---|---|
| REST API | Direct calls from control plane services |
| Inbound Webhooks | External systems push state updates |
| Event Bus | Messaging Provider subscription |
State Management¶
Transition State¶
StateTransitionResult result = echo.transitionState(
new StateKey("acme", "prod", "orders", "daily-ingest", "2024-01-15", "run-001"),
"RUNNING",
"pipeline-submitted"
);
Invalid Transitions¶
If a transition is not permitted by the state machine, Echo rejects it:
// Current state: COMPLETED
// Attempting: COMPLETED → RUNNING (not permitted)
StateTransitionResult result = echo.transitionState(key, "RUNNING", "retry");
// result.success() == false
// result.message() == "Invalid transition from COMPLETED to RUNNING"
Metrics¶
Persist¶
echo.persistMetric(new MetricsEnvelope(
"acme", "prod", "orders", "daily-ingest", "run-001",
ExecutionMode.BATCH, "spark-provider", "transform-stage",
"records.processed", 150000, CanonicalUnit.COUNT,
System.currentTimeMillis(), Map.of("partition", "2024-01-15")
));
Query¶
PaginatedResult<MetricsEnvelope> metrics = echo.queryMetrics(
new MetricsFilter("acme", "prod", "daily-ingest", "records.processed",
startTime, endTime, 100, null)
);
Lineage¶
Persist¶
echo.persistLineage(new LineageEvent(
"raw-orders", // source dataset
"enriched-orders", // destination dataset
List.of(new FieldMapping("order_id", "order_id", null),
new FieldMapping("amount", "total_amount", "SUM")),
"com.acme.transforms.OrderAggregation",
"daily-ingest", "run-001", "transform-stage",
Instant.now().toString(), "spark-provider", null
));
Query¶
LineageGraph graph = echo.queryLineage("enriched-orders", LineageDirection.BACKWARD);
// Returns upstream datasets and transformation relationships
Quality Results¶
echo.persistQualityResult(new QualityResult(
"orders", "daily-ingest", "run-001",
100000, 99500, 500, 0.995,
List.of(new RuleResult("not-null-check", 99800, 200, 0.998)),
Instant.now().toString()
));
Recovery Metadata¶
RecoveryMetadata recovery = echo.getRecoveryMetadata(stateKey);
// Returns: last checkpoint, failed stage, failure reason, retry count, recovery options
Outbound Webhooks¶
Subscribe to state change notifications:
String subscriptionId = echo.subscribe(new WebhookSubscription(
"https://alerts.acme.com/pipeline-events",
List.of("state.transition", "quality.breach"),
"acme", "prod", "shared-secret-123"
));
Data Contracts¶
Query current and historical data contracts: