From 52580059bf0bb1a8bc3db338996132be783f50e9 Mon Sep 17 00:00:00 2001 From: yahiaongh Date: Wed, 22 Jul 2026 11:37:01 +0100 Subject: [PATCH] test: close tracked gaps - detect() end-to-end tests with regression guard, EWMA baseline persistence for error_burst, mocked-pool unit tests for store.go write path --- services/anomaly-rust/src/detector/mod.rs | 161 +++++++++++++++++- services/ingestion-go/go.mod | 1 + services/ingestion-go/go.sum | 2 + .../ingestion-go/internal/migrate/migrate.go | 4 + services/ingestion-go/internal/store/store.go | 21 ++- .../ingestion-go/internal/store/store_test.go | 62 +++++++ 6 files changed, 244 insertions(+), 7 deletions(-) create mode 100644 services/ingestion-go/internal/store/store_test.go diff --git a/services/anomaly-rust/src/detector/mod.rs b/services/anomaly-rust/src/detector/mod.rs index 438f849..18027b9 100644 --- a/services/anomaly-rust/src/detector/mod.rs +++ b/services/anomaly-rust/src/detector/mod.rs @@ -49,6 +49,10 @@ pub struct ErrorRateBaseline { } impl ErrorRateBaseline { + /// Returns `(baseline_before_update, z_score, severity)` on anomaly, + /// or `None` if the current reading isn't anomalous. The baseline is + /// the EWMA value *before* this reading was folded in, so callers can + /// persist "what we expected" alongside "what we saw". pub fn update_and_check( &mut self, service: &str, @@ -56,7 +60,7 @@ impl ErrorRateBaseline { alpha: f64, warning: f64, critical: f64, - ) -> Option<(f64, &'static str)> { + ) -> Option<(f64, f64, &'static str)> { let baseline = *self.ewma.get(service).unwrap_or(¤t_error_rate); let deviation = current_error_rate - baseline; @@ -71,7 +75,7 @@ impl ErrorRateBaseline { let scale = baseline.max(0.01); // avoid division blowing up near zero let z = deviation / scale; - severity_for(z, warning, critical).map(|sev| (z, sev)) + severity_for(z, warning, critical).map(|sev| (baseline, z, sev)) } } @@ -161,7 +165,7 @@ pub fn detect( // suspicious drops toward zero are worth flagging) --- if current.event_count > 0 { let current_error_rate = current.error_count as f64 / current.event_count as f64; - if let Some((z, sev)) = error_baseline.update_and_check( + if let Some((baseline, z, sev)) = error_baseline.update_and_check( service, current_error_rate, cfg.ewma_alpha, @@ -173,7 +177,9 @@ pub fn detect( service: service.to_string(), anomaly_type: "error_burst".to_string(), metric_value: current_error_rate, - baseline_mean: 0.0, + baseline_mean: baseline, + // EWMA doesn't track variance, so there's no meaningful + // stddev to report here — 0.0 is accurate, not a placeholder. baseline_stddev: 0.0, z_score: z, severity: sev.to_string(), @@ -238,5 +244,152 @@ mod tests { // A sudden spike should be flagged against the now-stable low baseline. let result = baseline.update_and_check("checkout", 0.6, 0.3, 2.5, 4.0); assert!(result.is_some(), "expected spike to be flagged"); + let (baseline_value, _z, _sev) = result.unwrap(); + assert!( + (baseline_value - 0.02).abs() < 0.05, + "expected baseline near the primed 0.02 rate, got {baseline_value}" + ); + } + + fn make_bucket( + service: &str, + minute_offset: i64, + p95: f64, + max: f64, + errors: i64, + total: i64, + ) -> MinuteBucket { + MinuteBucket { + bucket: chrono::Utc::now() + chrono::Duration::minutes(minute_offset), + service: service.to_string(), + event_count: total, + error_count: errors, + avg_latency_ms: Some(p95 * 0.7), + p95_latency_ms: Some(p95), + max_latency_ms: Some(max), + } + } + + fn default_cfg() -> DetectionConfig { + DetectionConfig { + z_score_warning: 2.5, + z_score_critical: 4.0, + ewma_alpha: 0.3, + } + } + + #[test] + fn detect_flags_nothing_for_stable_service() { + let buckets = vec![ + make_bucket("checkout", -5, 100.0, 150.0, 1, 100), + make_bucket("checkout", -4, 105.0, 155.0, 1, 100), + make_bucket("checkout", -3, 98.0, 148.0, 1, 100), + make_bucket("checkout", -2, 102.0, 152.0, 1, 100), + make_bucket("checkout", -1, 101.0, 151.0, 1, 100), + ]; + let mut baseline = ErrorRateBaseline::default(); + let anomalies = detect(&buckets, &mut baseline, &default_cfg()); + assert!( + anomalies.is_empty(), + "expected no anomalies for a stable service, got: {anomalies:?}" + ); + } + + #[test] + fn detect_flags_latency_spike_but_not_a_speedup() { + // Five stable baseline minutes, then one minute that spikes p95 way up. + let mut buckets = vec![ + make_bucket("checkout", -6, 100.0, 150.0, 1, 100), + make_bucket("checkout", -5, 102.0, 152.0, 1, 100), + make_bucket("checkout", -4, 98.0, 148.0, 1, 100), + make_bucket("checkout", -3, 101.0, 151.0, 1, 100), + make_bucket("checkout", -2, 99.0, 149.0, 1, 100), + ]; + buckets.push(make_bucket("checkout", -1, 5000.0, 6000.0, 1, 100)); + + let mut baseline = ErrorRateBaseline::default(); + let anomalies = detect(&buckets, &mut baseline, &default_cfg()); + + let latency_anomalies: Vec<_> = anomalies + .iter() + .filter(|a| { + a.anomaly_type == "latency_spike" || a.anomaly_type == "max_latency_outlier" + }) + .collect(); + assert!( + !latency_anomalies.is_empty(), + "expected a latency spike to be detected" + ); + for a in &latency_anomalies { + assert!( + a.z_score > 0.0, + "latency anomaly must have a positive z_score, got {} for {}", + a.z_score, + a.anomaly_type + ); + } + + // Regression guard for the original bug: a service getting FASTER + // than baseline must never be flagged. + let mut faster_buckets = vec![ + make_bucket("payments", -6, 500.0, 600.0, 1, 100), + make_bucket("payments", -5, 510.0, 610.0, 1, 100), + make_bucket("payments", -4, 490.0, 590.0, 1, 100), + make_bucket("payments", -3, 505.0, 605.0, 1, 100), + make_bucket("payments", -2, 495.0, 595.0, 1, 100), + ]; + faster_buckets.push(make_bucket("payments", -1, 10.0, 15.0, 1, 100)); + let mut baseline2 = ErrorRateBaseline::default(); + let anomalies2 = detect(&faster_buckets, &mut baseline2, &default_cfg()); + let false_positives: Vec<_> = anomalies2 + .iter() + .filter(|a| { + a.anomaly_type == "latency_spike" || a.anomaly_type == "max_latency_outlier" + }) + .collect(); + assert!( + false_positives.is_empty(), + "a service running FASTER than baseline must not be flagged, got: {false_positives:?}" + ); + } + + #[test] + fn detect_flags_error_burst() { + let mut buckets = vec![]; + for i in (-6..-1).rev() { + buckets.push(make_bucket("inventory", i, 100.0, 150.0, 2, 100)); + } + buckets.push(make_bucket("inventory", -1, 100.0, 150.0, 60, 100)); + + let mut baseline = ErrorRateBaseline::default(); + // Prime the EWMA baseline the same way main.rs would across prior cycles. + for i in 0..5 { + baseline.update_and_check("inventory", 0.02, 0.3, 2.5, 4.0); + let _ = i; + } + + let anomalies = detect(&buckets, &mut baseline, &default_cfg()); + let error_anomalies: Vec<_> = anomalies + .iter() + .filter(|a| a.anomaly_type == "error_burst") + .collect(); + assert!( + !error_anomalies.is_empty(), + "expected an error burst to be detected, got: {anomalies:?}" + ); + } + + #[test] + fn detect_skips_services_with_insufficient_history() { + let buckets = vec![ + make_bucket("new-service", -1, 100.0, 150.0, 1, 100), + make_bucket("new-service", 0, 5000.0, 6000.0, 50, 100), + ]; + let mut baseline = ErrorRateBaseline::default(); + let anomalies = detect(&buckets, &mut baseline, &default_cfg()); + assert!( + anomalies.is_empty(), + "expected no anomalies for a service with fewer than 3 buckets of history" + ); } } diff --git a/services/ingestion-go/go.mod b/services/ingestion-go/go.mod index 9ac9ade..cfd30f8 100644 --- a/services/ingestion-go/go.mod +++ b/services/ingestion-go/go.mod @@ -17,6 +17,7 @@ require ( github.com/jackc/puddle/v2 v2.2.2 // indirect github.com/klauspost/compress v1.17.9 // indirect github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect + github.com/pashagolub/pgxmock/v4 v4.3.0 // indirect github.com/pierrec/lz4/v4 v4.1.15 // indirect github.com/prometheus/client_model v0.6.1 // indirect github.com/prometheus/common v0.55.0 // indirect diff --git a/services/ingestion-go/go.sum b/services/ingestion-go/go.sum index af4bd03..d840541 100644 --- a/services/ingestion-go/go.sum +++ b/services/ingestion-go/go.sum @@ -24,6 +24,8 @@ github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0 github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw= github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= +github.com/pashagolub/pgxmock/v4 v4.3.0 h1:DqT7fk0OCK6H0GvqtcMsLpv8cIwWqdxWgfZNLeHCb/s= +github.com/pashagolub/pgxmock/v4 v4.3.0/go.mod h1:9VoVHXwS3XR/yPtKGzwQvwZX1kzGB9sM8SviDcHDa3A= github.com/pierrec/lz4/v4 v4.1.15 h1:MO0/ucJhngq7299dKLwIMtgTfbkoSPF6AoMYDd8Q4q0= github.com/pierrec/lz4/v4 v4.1.15/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= diff --git a/services/ingestion-go/internal/migrate/migrate.go b/services/ingestion-go/internal/migrate/migrate.go index a380c3b..b39358a 100644 --- a/services/ingestion-go/internal/migrate/migrate.go +++ b/services/ingestion-go/internal/migrate/migrate.go @@ -15,6 +15,10 @@ var initSchema string // It is idempotent: every statement uses IF NOT EXISTS / ON CONFLICT-style // guards, so it's safe to run on every service startup, whether the schema // already exists or the volume is brand new. +// +// Takes a concrete *pgxpool.Pool (rather than store.PgxPool) specifically to +// avoid an import cycle: store's integration tests import migrate to apply +// the schema, so migrate cannot import store back. func Apply(ctx context.Context, pool *pgxpool.Pool, logger *slog.Logger) error { logger.Info("applying database schema migration") if _, err := pool.Exec(ctx, initSchema); err != nil { diff --git a/services/ingestion-go/internal/store/store.go b/services/ingestion-go/internal/store/store.go index 9ecb955..1399717 100644 --- a/services/ingestion-go/internal/store/store.go +++ b/services/ingestion-go/internal/store/store.go @@ -7,13 +7,25 @@ import ( "time" "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" "github.com/jackc/pgx/v5/pgxpool" "github.com/yahiaongh/sentinelops/services/ingestion-go/internal/consumer" "github.com/yahiaongh/sentinelops/services/ingestion-go/internal/metrics" ) +// PgxPool is the subset of *pgxpool.Pool's behavior that Store depends on. +// Defined as an interface (rather than using *pgxpool.Pool directly) so +// tests can substitute a mock pool without needing a live database. +type PgxPool interface { + Ping(ctx context.Context) error + Begin(ctx context.Context) (pgx.Tx, error) + Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error) + QueryRow(ctx context.Context, sql string, args ...any) pgx.Row + Close() +} + type Store struct { - pool *pgxpool.Pool + pool PgxPool logger *slog.Logger } @@ -33,9 +45,12 @@ func (s *Store) Close() { } // Pool exposes the underlying connection pool for use by other internal -// packages (e.g. migrate) that need direct database access. +// packages (e.g. migrate) that need direct database access. Returns the +// concrete *pgxpool.Pool (not the PgxPool interface) since callers like +// migrate.Apply need the real type, and s.pool is always a genuine +// *pgxpool.Pool outside of unit tests. func (s *Store) Pool() *pgxpool.Pool { - return s.pool + return s.pool.(*pgxpool.Pool) } const insertLogEventQuery = ` diff --git a/services/ingestion-go/internal/store/store_test.go b/services/ingestion-go/internal/store/store_test.go new file mode 100644 index 0000000..f2ef65a --- /dev/null +++ b/services/ingestion-go/internal/store/store_test.go @@ -0,0 +1,62 @@ +package store + +import ( + "context" + "log/slog" + "os" + "testing" + + "github.com/pashagolub/pgxmock/v4" + "github.com/yahiaongh/sentinelops/services/ingestion-go/internal/consumer" +) + +func testLogger() *slog.Logger { + return slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelError})) +} + +func TestFlushBatch_EmptySlice_NoOp(t *testing.T) { + mock, err := pgxmock.NewPool() + if err != nil { + t.Fatalf("failed to create mock pool: %v", err) + } + defer mock.Close() + + s := &Store{pool: mock, logger: testLogger()} + + // No expectations set on the mock — if FlushBatch tries to touch the + // database for an empty slice, ExpectationsWereMet will fail below, + // proving the early-return short-circuit actually works. + if err := s.FlushBatch(context.Background(), nil); err != nil { + t.Errorf("expected nil error for empty slice, got: %v", err) + } + + if err := mock.ExpectationsWereMet(); err != nil { + t.Errorf("expected zero database interaction for an empty batch, got: %v", err) + } +} + +func TestFlushBatch_AllInvalidTimestamps_NoOp(t *testing.T) { + mock, err := pgxmock.NewPool() + if err != nil { + t.Fatalf("failed to create mock pool: %v", err) + } + defer mock.Close() + + s := &Store{pool: mock, logger: testLogger()} + + events := []consumer.LogEvent{ + {EventID: "a", Timestamp: "not-a-timestamp"}, + {EventID: "b", Timestamp: ""}, + } + + // Every event has an unparsable timestamp, so FlushBatch should skip + // all of them and never open a transaction — again proven by the mock + // having zero expectations set. + if err := s.FlushBatch(context.Background(), events); err != nil { + t.Errorf("expected nil error when all events have invalid timestamps, got: %v", err) + } + + if err := mock.ExpectationsWereMet(); err != nil { + t.Errorf("expected zero database interaction when no events are valid, got: %v", err) + } +}