ETL Engine Observability

Overview

The observability stack provides comprehensive monitoring for pipeline execution:

• Run logs - Complete execution history with timing and status
• Data lineage - Column-level tracking of data flow
• Metrics - Aggregated KPIs with cost modeling
• Alerts - Rule-based alerting with silence windows
• Audit - Tamper-evident audit trail
• Dashboard API - Query interface for dashboards
• Field redaction - Automatic PII masking in logs

ObservabilityStore

Central storage for all observability data:

public class ObservabilityStore : ILineageStore
{
    public ObservabilityStore();

    // Run logs
    public Task SaveRunLogAsync(PipelineRunLog log);
    public Task<PipelineRunLog?> GetRunLogAsync(string runId);
    public Task<IReadOnlyList<PipelineRunLog>> QueryRunLogsAsync(RunLogQuery query);

    // Lineage
    public Task AppendLineageAsync(IEnumerable<DataLineageRecord> records);
    public Task<IReadOnlyList<DataLineageRecord>> GetLineageAsync(string runId);

    // Alerts
    public Task AppendAlertEventAsync(AlertEvent evt);
    public Task<IReadOnlyList<AlertEvent>> GetAlertEventsAsync(AlertEventQuery query);
    public Task UpdateAlertAcknowledgementAsync(string eventId, string by, DateTime at);

    // Audit
    public Task AppendAuditAsync(AuditEntry entry);
    public Task<int> VerifyAuditIntegrityAsync();
}

Storage Locations

MetricsEngine

Computes aggregated KPIs from run logs:

public class MetricsEngine
{
    public MetricsEngine(ObservabilityStore store);
    public MetricsEngine(ObservabilityStore store, CostModel costModel);

    // Historical metrics
    public Task<PipelineMetrics> ComputeAsync(
        string pipelineId, DateTime from, DateTime to);

    public Task<IReadOnlyList<PipelineMetrics>> ComputeAllAsync(
        DateTime from, DateTime to);

    public Task<IReadOnlyList<PipelineMetrics>> ComputeByClassAsync(
        DateTime from, DateTime to);

    public Task<IReadOnlyList<MigrationKpiSnapshot>> ComputeByTierAsync(
        DateTime from, DateTime to);

    // Live metrics
    public void RecordStart(string runId, string pipelineId, string pipelineName);
    public void RecordStart(string runId, string pipelineId, string pipelineName, string workloadClass);
    public void RecordCompletion(string runId);
    public void RecordFailure(string runId, Exception ex);
    public void RecordRecordsProcessed(string runId, long count);
    public void RecordRecordsRejected(string runId, long count);
}

PipelineMetrics

AlertingEngine

Rule-based alerting with silence windows:

public class AlertingEngine
{
    public event EventHandler<AlertEvent>? AlertFired;

    public AlertingEngine(
        ObservabilityStore store,
        Func<string, IPipelineNotifier?> notifierFactory);

    // Rule management
    public void AddRule(AlertRule rule);
    public void RemoveRule(string ruleId);
    public IReadOnlyList<AlertRule> GetRules();

    // Evaluation
    public Task EvaluateAsync(PipelineRunLog runLog, CancellationToken token);
    public Task EvaluateLivenessAsync(CancellationToken token);
    public Task AcknowledgeAsync(string eventId, string acknowledgedBy);
}

AlertRule

AlertEvent

PipelineDashboardApi

Query interface for building dashboards:

public class PipelineDashboardApi
{
    public PipelineDashboardApi(
        ObservabilityStore store,
        MetricsEngine metrics,
        AlertingEngine alerting);

    // Summary
    public Task<DashboardSummary> GetSummaryAsync(DateTime from, DateTime to);

    // Pipeline metrics
    public Task<PipelineMetrics> GetPipelineMetricsAsync(
        string pipelineId, DateTime from, DateTime to);

    // Active runs
    public IReadOnlyList<LiveRunStatus> GetActiveRuns();

    // Alerts
    public Task<IReadOnlyList<AlertEvent>> GetRecentAlertsAsync(int limit = 100);
    public Task<IReadOnlyList<AlertEvent>> GetUnacknowledgedAlertsAsync();

    // Run history
    public Task<IReadOnlyList<PipelineRunLog>> GetRunHistoryAsync(
        string pipelineId, int limit = 50);

    // Lineage
    public Task<LineageReport> GetLineageAsync(string runId);
}

DashboardSummary

Core Data Types

PipelineRunLog

public class PipelineRunLog
{
    public string RunId { get; set; }
    public string PipelineId { get; set; }
    public string PipelineName { get; set; }
    public string WorkloadClass { get; set; }
    public RunStatus Status { get; set; }
    public DateTime StartedAt { get; set; }
    public DateTime? CompletedAt { get; set; }
    public TimeSpan? Duration { get; set; }
    public long RecordsProcessed { get; set; }
    public long RecordsRejected { get; set; }
    public long RecordsFailed { get; set; }
    public string? ErrorMessage { get; set; }
    public string? ErrorDetail { get; set; }
    public Dictionary<string, object> Parameters { get; set; }
    public List<PipelineStepLog> StepLogs { get; set; }
}

DataLineageRecord

public class DataLineageRecord
{
    public string RunId { get; set; }
    public string RecordId { get; set; }
    public string SourceEntity { get; set; }
    public string SourceField { get; set; }
    public string Transformation { get; set; }
    public string DestinationEntity { get; set; }
    public string DestinationField { get; set; }
    public DateTime Timestamp { get; set; }
}

AuditEntry

public class AuditEntry
{
    public string EntryId { get; set; }
    public string Action { get; set; }
    public string Actor { get; set; }
    public string Resource { get; set; }
    public string Details { get; set; }
    public DateTime Timestamp { get; set; }
    public string Hash { get; set; } // Tamper-evident
}

Usage Examples

Setting Up Observability

var store = new ObservabilityStore();
var metrics = new MetricsEngine(store);
var alerting = new AlertingEngine(store, notifierId =>
{
    if (notifierId == "email") return new EmailNotifier();
    if (notifierId == "slack") return new SlackNotifier();
    return null;
});

var dashboard = new PipelineDashboardApi(store, metrics, alerting);

// Attach to pipeline manager
manager.ObservabilityStore = store;
manager.Metrics = metrics;
manager.Alerting = alerting;

Adding Alert Rules

// Alert on failure
alerting.AddRule(new AlertRule
{
    Name = "Pipeline Failure",
    Trigger = AlertTrigger.OnFailure,
    Severity = AlertSeverity.Error,
    SilenceMinutes = 30,
    NotifierIds = new[] { "email", "slack" }
});

// Alert on slow runs
alerting.AddRule(new AlertRule
{
    Name = "Slow Pipeline",
    Trigger = AlertTrigger.OnCondition,
    Condition = "Duration > 300", // 5 minutes
    Severity = AlertSeverity.Warning,
    PipelineIds = new[] { "customer-sync" },
    SilenceMinutes = 60
});

// Alert on missing runs
alerting.AddRule(new AlertRule
{
    Name = "Missing Run",
    Trigger = AlertTrigger.OnNoRunWithin,
    Condition = "HoursWithoutRun > 2",
    Severity = AlertSeverity.Critical,
    PipelineIds = new[] { "critical-pipeline" },
    NotifierIds = new[] { "pagerduty" }
});

Querying Metrics

// Pipeline metrics
var metrics = await dashboard.GetPipelineMetricsAsync(
    "customer-sync",
    from: DateTime.UtcNow.AddDays(-7),
    to: DateTime.UtcNow);

Console.WriteLine($"Success rate: {metrics.SuccessRate:F1}%");
Console.WriteLine($"Avg duration: {metrics.AverageDuration}");
Console.WriteLine($"Records processed: {metrics.TotalRecordsProcessed:N0}");

// Dashboard summary
var summary = await dashboard.GetSummaryAsync(
    DateTime.UtcNow.AddDays(-1),
    DateTime.UtcNow);

Console.WriteLine($"Active runs: {summary.ActiveRuns}");
Console.WriteLine($"Active alerts: {summary.ActiveAlerts}");

Monitoring Active Runs

var active = dashboard.GetActiveRuns();
foreach (var run in active)
{
    Console.WriteLine($"{run.PipelineName}: {run.Progress}% " +
        $"({run.RecordsProcessed:N0} records)");
}

Audit Trail

// Log audit event
await store.AppendAuditAsync(new AuditEntry
{
    Action = "PipelineModified",
    Actor = "admin",
    Resource = "customer-sync",
    Details = "Changed batch size from 100 to 500",
    Timestamp = DateTime.UtcNow
});

// Verify integrity
var tamperedCount = await store.VerifyAuditIntegrityAsync();
if (tamperedCount > 0)
    Console.WriteLine($"WARNING: {tamperedCount} tampered audit entries detected!");