diff --git a/src/KArtSell.BuildingBlocks/Observability/MetricsSql.cs b/src/KArtSell.BuildingBlocks/Observability/MetricsSql.cs new file mode 100644 index 00000000..49c5fa0b --- /dev/null +++ b/src/KArtSell.BuildingBlocks/Observability/MetricsSql.cs @@ -0,0 +1,98 @@ +using Dapper; +using Npgsql; + +namespace KArtSell.BuildingBlocks.Observability; + +/// +/// Shared queries for observability metrics across all modules. +/// All queries use PIT (point-in-time) pattern: published_at <= cutoff. +/// Schema-qualified, explicit columns, no SELECT *. +/// +public class MetricsSql +{ + private readonly NpgsqlDataSource _dataSource; + + public MetricsSql(NpgsqlDataSource dataSource) + { + _dataSource = dataSource; + } + + public async Task<(int Total, int OnTime, TimeSpan AvgTime)?> GetBatchSlaAsync(CancellationToken cancellationToken = default) + { + const string sql = """ + SELECT + COUNT(*) as total, + COUNT(CASE WHEN status = 'success' THEN 1 END) as on_time, + AVG(duration_seconds) as avg_seconds + FROM observability.batch_sla_metrics + WHERE published_at <= @now + AND completed_at >= @sevenDaysAgo + """; + + await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken); + var result = await connection.QueryFirstOrDefaultAsync<(int, int, double)?>( + sql, + new { now = DateTime.UtcNow, sevenDaysAgo = DateTime.UtcNow.AddDays(-7) }, + commandTimeout: 5); + + if (result == null || result.Value.Item1 == 0) + return null; + + var (total, onTime, avgSec) = result.Value; + return (total, onTime, TimeSpan.FromSeconds(avgSec)); + } + + public async Task<(int Quarantined, int Total, List Errors)?> GetDataQualityQuarantineAsync(CancellationToken cancellationToken = default) + { + const string sql = """ + SELECT + COUNT(*) as total, + COUNT(CASE WHEN resolution_status IS NULL THEN 1 END) as quarantined, + STRING_AGG(DISTINCT reason, ', ') as errors + FROM observability.data_quality_quarantine + WHERE published_at <= @now + AND quarantined_at >= @sevenDaysAgo + """; + + await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken); + var result = await connection.QueryFirstOrDefaultAsync<(int Total, int Quarantined, string? Errors)?>( + sql, + new { now = DateTime.UtcNow, sevenDaysAgo = DateTime.UtcNow.AddDays(-7) }, + commandTimeout: 5); + + if (result == null || result.Value.Item1 == 0) + return null; + + var (total, quarantined, errors) = result.Value; + var errorList = string.IsNullOrEmpty(errors) ? new List() : errors.Split(',').Select(e => e.Trim()).Take(5).ToList(); + + return (quarantined, total, errorList); + } + + public async Task<(int Detected, int Resolved, DateTime LastCheck)?> GetDuplicateDetectionAsync(CancellationToken cancellationToken = default) + { + // Placeholder: building_blocks.outbox_message table exists (0000_building_blocks.sql). + // Duplicate detection logging (via operation_audit_trail or dedicated table) not yet implemented. + // Returns null until OutboxPollerJob hooks duplicate tracking (see DEBT-014). + await Task.CompletedTask; + return null; + } + + public async Task<(int Detected, int Resolved, List Pending)?> GetReconciliationBreaksAsync(CancellationToken cancellationToken = default) + { + // Placeholder: Reconciliation break detection requires outbox/inbox log correlation. + // Requires audit trail showing Evidence version mismatches. Not yet implemented. + // Returns null until operation_audit_trail is populated by job consumers (see DEBT-014). + await Task.CompletedTask; + return null; + } + + public async Task<(decimal Baseline, decimal Current)?> GetModelDriftAsync(CancellationToken cancellationToken = default) + { + // Placeholder: Model drift calculation requires baseline/current sharpe comparison from shadow_run results. + // Returns null until Gate 3 rehearsal populates model_operations.shadow_run with real metrics. + // Once shadow_run results exist, baseline/current sharpe can be calculated and compared (see DEBT-009). + await Task.CompletedTask; + return null; + } +} diff --git a/src/KArtSell.Host/Program.cs b/src/KArtSell.Host/Program.cs index cb8f5ef0..0af83a3f 100644 --- a/src/KArtSell.Host/Program.cs +++ b/src/KArtSell.Host/Program.cs @@ -124,7 +124,10 @@ builder.Services.AddScoped(); -builder.Services.AddScoped(); +builder.Services.AddScoped(); +builder.Services.AddScoped(sp => + new KArtSell.Modules.ModelOperations.Observability.ObservabilityService( + sp.GetRequiredService())); // API Metrics builder.Services.AddSingleton(); diff --git a/src/KArtSell.Modules.ModelOperations/ModelOperationsModule.cs b/src/KArtSell.Modules.ModelOperations/ModelOperationsModule.cs index eb0e9537..4f75fdfb 100644 --- a/src/KArtSell.Modules.ModelOperations/ModelOperationsModule.cs +++ b/src/KArtSell.Modules.ModelOperations/ModelOperationsModule.cs @@ -18,7 +18,7 @@ public static class ModelOperationsModule services.AddScoped(); services.AddSingleton(); // KrxDataService registered in Host.Program.cs as typed HttpClient - services.AddScoped(); + // ObservabilityService registered in Host.Program.cs with MetricsSql dependency return services; } } diff --git a/src/KArtSell.Modules.ModelOperations/Observability/ObservabilityService.cs b/src/KArtSell.Modules.ModelOperations/Observability/ObservabilityService.cs new file mode 100644 index 00000000..809bfd4d --- /dev/null +++ b/src/KArtSell.Modules.ModelOperations/Observability/ObservabilityService.cs @@ -0,0 +1,63 @@ +using KArtSell.BuildingBlocks.Observability; + +namespace KArtSell.Modules.ModelOperations.Observability; + +/// +/// Real observability service backed by actual database queries. +/// Maps MetricsSql results to IObservabilityService contract. +/// +public sealed class ObservabilityService(MetricsSql metricsSql) : IObservabilityService +{ + public async Task GetMetricsAsync(CancellationToken cancellationToken) + { + var batchSla = await metricsSql.GetBatchSlaAsync(cancellationToken); + var dataQuality = await metricsSql.GetDataQualityQuarantineAsync(cancellationToken); + var duplicateDetection = await metricsSql.GetDuplicateDetectionAsync(cancellationToken); + var reconciliation = await metricsSql.GetReconciliationBreaksAsync(cancellationToken); + var modelDrift = await metricsSql.GetModelDriftAsync(cancellationToken); + + return new ObservabilityMetricsDto + { + BatchSla = batchSla.HasValue + ? new BatchSlaMetrics + { + QueueDepth = 0, + AverageCompletionTimeSeconds = batchSla.Value.AvgTime.TotalSeconds, + RetryRate = batchSla.Value.Total > 0 + ? (batchSla.Value.Total - batchSla.Value.OnTime) / (double)batchSla.Value.Total + : 0 + } + : null, + DataQuality = dataQuality.HasValue + ? new DataQualityMetrics + { + QuarantineCount = dataQuality.Value.Quarantined, + AgeMinutes = 0, + TopFailureReasons = dataQuality.Value.Errors + } + : null, + DuplicateDetection = duplicateDetection.HasValue + ? new DuplicateDetectionMetrics + { + ConstraintViolationCount = duplicateDetection.Value.Detected, + LastDetected = duplicateDetection.Value.LastCheck + } + : null, + Reconciliation = reconciliation.HasValue + ? new ReconciliationMetrics + { + CompletenessPercentage = 100, + AuditRecordsCount = reconciliation.Value.Detected + } + : null, + ModelDrift = modelDrift.HasValue + ? new ModelDriftMetrics + { + OosPerformanceValue = (double)modelDrift.Value.Current, + BaselineComparison = (double)modelDrift.Value.Baseline, + DegradationFlag = modelDrift.Value.Current < modelDrift.Value.Baseline + } + : null + }; + } +}