diff --git a/backend/internal/hub/db/migrations/000036_create_application_health_changes_table.down.sql b/backend/internal/hub/db/migrations/000036_create_application_health_changes_table.down.sql new file mode 100644 index 00000000..60d854c6 --- /dev/null +++ b/backend/internal/hub/db/migrations/000036_create_application_health_changes_table.down.sql @@ -0,0 +1,3 @@ +DROP INDEX IF EXISTS idx_application_health_changes_changed; +DROP INDEX IF EXISTS idx_application_health_changes_app_changed; +DROP TABLE IF EXISTS application_health_changes; diff --git a/backend/internal/hub/db/migrations/000036_create_application_health_changes_table.up.sql b/backend/internal/hub/db/migrations/000036_create_application_health_changes_table.up.sql new file mode 100644 index 00000000..51723055 --- /dev/null +++ b/backend/internal/hub/db/migrations/000036_create_application_health_changes_table.up.sql @@ -0,0 +1,14 @@ +CREATE TABLE IF NOT EXISTS application_health_changes ( + id TEXT PRIMARY KEY, + application_id TEXT NOT NULL, + status TEXT NOT NULL, + reason TEXT, + services TEXT, + changed_at DATETIME NOT NULL DEFAULT (datetime('now')), + FOREIGN KEY (application_id) REFERENCES applications(id) ON DELETE CASCADE +); + +CREATE INDEX IF NOT EXISTS idx_application_health_changes_app_changed + ON application_health_changes(application_id, changed_at DESC, id DESC); +CREATE INDEX IF NOT EXISTS idx_application_health_changes_changed + ON application_health_changes(changed_at); diff --git a/backend/internal/hub/healthhistory/healthhistory.go b/backend/internal/hub/healthhistory/healthhistory.go new file mode 100644 index 00000000..fc0c3b76 --- /dev/null +++ b/backend/internal/hub/healthhistory/healthhistory.go @@ -0,0 +1,179 @@ +// Package healthhistory records application health changes and keeps them +// for the availability statistics. +package healthhistory + +import ( + "context" + "errors" + "strings" + "sync" + "time" + "unicode/utf8" + + "github.com/OrcaCD/orca-cd/internal/hub/db" + "github.com/OrcaCD/orca-cd/internal/hub/models" + "github.com/rs/zerolog" + "gorm.io/gorm" +) + +// Retention is how long health changes are kept. The longest statistics range +// must fit into it. +const Retention = 30 * 24 * time.Hour + +const ( + pruneInterval = 24 * time.Hour + // Services and their names come from agents, so both are bounded before + // they are persisted. + maxServices = 32 + maxServiceNameLength = 128 +) + +// Record stores a health change unless the application's latest recorded +// status is the same. A change dated before the latest one is moved to the +// latest change's time so the history stays ordered. The very first change of +// an application is not recorded when it is unknown, as no data is implied +// before the first change anyway. +// +// Callers must serialize changes per application; the hub does so by +// handling each agent's messages sequentially. +func Record(ctx context.Context, change models.ApplicationHealthChange) (bool, error) { + if change.ChangedAt.IsZero() { + change.ChangedAt = time.Now() + } + + latest, err := Latest(ctx, change.ApplicationId) + switch { + case errors.Is(err, gorm.ErrRecordNotFound): + if change.Status == models.UnknownHealth { + return false, nil + } + case err != nil: + return false, err + case latest.Status == change.Status: + return false, nil + case change.ChangedAt.Before(latest.ChangedAt): + change.ChangedAt = latest.ChangedAt + } + + change.Services = sanitizeServices(change.Services) + if err := gorm.G[models.ApplicationHealthChange](db.DB).Create(ctx, &change); err != nil { + return false, err + } + return true, nil +} + +// Latest returns the most recent health change of an application. +func Latest(ctx context.Context, applicationID string) (models.ApplicationHealthChange, error) { + return gorm.G[models.ApplicationHealthChange](db.DB). + Where("application_id = ?", applicationID). + Order("changed_at DESC, id DESC"). + First(ctx) +} + +// RecordAgentOffline marks the health of all applications of an agent as +// unknown because the agent is no longer connected. +func RecordAgentOffline(ctx context.Context, agentID string, at time.Time) error { + apps, err := gorm.G[models.Application](db.DB).Select("id").Where("agent_id = ?", agentID).Find(ctx) + if err != nil { + return err + } + + reason := models.HealthChangeReasonAgentOffline + var errs []error + for _, app := range apps { + if _, err := Record(ctx, models.ApplicationHealthChange{ + ApplicationId: app.Id, + Status: models.UnknownHealth, + Reason: &reason, + ChangedAt: at, + }); err != nil { + errs = append(errs, err) + } + } + return errors.Join(errs...) +} + +// Prune deletes health changes older than the retention. The latest change +// before the cutoff is kept per application, as it defines the status at the +// start of the longest range. +func Prune(ctx context.Context, now time.Time) (int, error) { + cutoff := now.Add(-Retention).UTC() + return gorm.G[models.ApplicationHealthChange](db.DB). + Where(`changed_at < ? AND id NOT IN ( + SELECT id FROM ( + SELECT id, ROW_NUMBER() OVER (PARTITION BY application_id ORDER BY changed_at DESC, id DESC) AS position + FROM application_health_changes + WHERE changed_at < ? + ) WHERE position = 1 + )`, cutoff, cutoff). + Delete(ctx) +} + +// StartPruner prunes the health history now and then once a day. The returned +// stop function cancels a running prune and waits for the pruner to exit, so +// the database can be closed afterwards. +func StartPruner(log *zerolog.Logger) (stop func()) { + ctx, cancel := context.WithCancel(context.Background()) + stopped := make(chan struct{}) + prune := func() { + pruneCtx, cancelPrune := context.WithTimeout(ctx, time.Minute) + defer cancelPrune() + deleted, err := Prune(pruneCtx, time.Now()) + if err != nil { + if ctx.Err() == nil { + log.Error().Err(err).Msg("failed to prune application health history") + } + return + } + if deleted > 0 { + log.Debug().Int("count", deleted).Msg("pruned application health history") + } + } + + go func() { + defer close(stopped) + prune() + ticker := time.NewTicker(pruneInterval) + defer ticker.Stop() + for { + select { + case <-ticker.C: + prune() + case <-ctx.Done(): + return + } + } + }() + var once sync.Once + return func() { + once.Do(func() { + cancel() + <-stopped + }) + } +} + +func sanitizeServices(services []models.ServiceHealth) []models.ServiceHealth { + if len(services) == 0 { + return nil + } + services = services[:min(len(services), maxServices)] + sanitized := make([]models.ServiceHealth, 0, len(services)) + for _, service := range services { + name := strings.ToValidUTF8(service.Name, "") + name = strings.TrimSpace(truncate(strings.TrimSpace(name), maxServiceNameLength)) + sanitized = append(sanitized, models.ServiceHealth{Name: name, Status: service.Status}) + } + return sanitized +} + +// truncate cuts s to at most maxBytes without splitting a UTF-8 character. +func truncate(s string, maxBytes int) string { + if len(s) <= maxBytes { + return s + } + for maxBytes > 0 && !utf8.RuneStart(s[maxBytes]) { + maxBytes-- + } + return s[:maxBytes] +} diff --git a/backend/internal/hub/healthhistory/healthhistory_test.go b/backend/internal/hub/healthhistory/healthhistory_test.go new file mode 100644 index 00000000..a87ad04d --- /dev/null +++ b/backend/internal/hub/healthhistory/healthhistory_test.go @@ -0,0 +1,309 @@ +package healthhistory + +import ( + "context" + "path/filepath" + "slices" + "strings" + "testing" + "time" + + "github.com/OrcaCD/orca-cd/internal/hub/crypto" + "github.com/OrcaCD/orca-cd/internal/hub/db" + "github.com/OrcaCD/orca-cd/internal/hub/models" + "github.com/rs/zerolog" + "gorm.io/driver/sqlite" + "gorm.io/gorm" + gormlogger "gorm.io/gorm/logger" +) + +const testRepositoryID = "repository-1" + +// setupTestDB seeds the agents agent-1 and agent-2 that applications can be +// assigned to. +func setupTestDB(t *testing.T) { + t.Helper() + + testDB, err := gorm.Open(sqlite.Open(filepath.Join(t.TempDir(), "health.db")+"?_foreign_keys=on"), &gorm.Config{ + Logger: gormlogger.Discard, + }) + if err != nil { + t.Fatalf("open test db: %v", err) + } + if err := testDB.AutoMigrate(&models.User{}, &models.Repository{}, &models.Agent{}, &models.Application{}, &models.ApplicationHealthChange{}); err != nil { + t.Fatalf("migrate test db: %v", err) + } + sqlDB, err := testDB.DB() + if err != nil { + t.Fatalf("get sql db: %v", err) + } + previousDB := db.DB + db.DB = testDB + t.Cleanup(func() { + db.DB = previousDB + _ = sqlDB.Close() + }) + + if err := crypto.Init("test-secret-that-is-long-enough-32chars"); err != nil { + t.Fatalf("init crypto: %v", err) + } + + ctx := t.Context() + if err := gorm.G[models.User](testDB).Create(ctx, &models.User{Id: "user-1", Email: "alex@example.com", Name: "Alex", Role: models.UserRoleAdmin}); err != nil { + t.Fatalf("seed user: %v", err) + } + if err := gorm.G[models.Repository](testDB).Create(ctx, &models.Repository{ + Id: testRepositoryID, + Name: "owner/repository", + Url: "https://example.com/owner/repository.git", + Provider: models.Generic, + AuthMethod: models.AuthMethodNone, + SyncType: models.SyncTypeManual, + SyncStatus: models.SyncStatusUnknown, + CreatedBy: "user-1", + }); err != nil { + t.Fatalf("seed repository: %v", err) + } + for _, id := range []string{"agent-1", "agent-2"} { + if err := gorm.G[models.Agent](testDB).Create(ctx, &models.Agent{ + Id: id, + Name: crypto.EncryptedString(id), + KeyId: crypto.EncryptedString("key-" + id), + }); err != nil { + t.Fatalf("seed agent: %v", err) + } + } +} + +func seedApp(t *testing.T, agentID string) models.Application { + t.Helper() + app := models.Application{ + Name: crypto.EncryptedString("app"), + RepositoryId: testRepositoryID, + AgentId: agentID, + SyncStatus: models.Synced, + HealthStatus: models.Healthy, + Branch: "main", + Path: "compose.yaml", + ComposeFile: crypto.EncryptedString("services: {}\n"), + } + if err := db.DB.Select("*").Create(&app).Error; err != nil { + t.Fatalf("seed application: %v", err) + } + return app +} + +func changesOf(t *testing.T, appID string) []models.ApplicationHealthChange { + t.Helper() + changes, err := gorm.G[models.ApplicationHealthChange](db.DB). + Where("application_id = ?", appID). + Order("changed_at, id"). + Find(context.Background()) + if err != nil { + t.Fatalf("load health changes: %v", err) + } + return changes +} + +func record(t *testing.T, change models.ApplicationHealthChange) bool { + t.Helper() + recorded, err := Record(t.Context(), change) + if err != nil { + t.Fatalf("Record() error: %v", err) + } + return recorded +} + +func TestRecord_OnlyStoresChanges(t *testing.T) { + setupTestDB(t) + app := seedApp(t, "agent-1") + + if record(t, models.ApplicationHealthChange{ApplicationId: app.Id, Status: models.UnknownHealth}) { + t.Error("an unknown first status must not be recorded") + } + if !record(t, models.ApplicationHealthChange{ApplicationId: app.Id, Status: models.Healthy}) { + t.Error("expected the first known status to be recorded") + } + if record(t, models.ApplicationHealthChange{ApplicationId: app.Id, Status: models.Healthy}) { + t.Error("an unchanged status must not be recorded") + } + if !record(t, models.ApplicationHealthChange{ApplicationId: app.Id, Status: models.Degraded, Services: []models.ServiceHealth{{Name: "web", Status: models.Unhealthy}}}) { + t.Error("expected the degraded status to be recorded") + } + if !record(t, models.ApplicationHealthChange{ApplicationId: app.Id, Status: models.UnknownHealth}) { + t.Error("expected unknown after a known status to be recorded") + } + + changes := changesOf(t, app.Id) + if len(changes) != 3 { + t.Fatalf("expected 3 changes, got %d", len(changes)) + } + if changes[1].Status != models.Degraded || len(changes[1].Services) != 1 || changes[1].Services[0].Name != "web" { + t.Errorf("expected degraded change with failing service web, got %+v", changes[1]) + } + if changes[2].Status != models.UnknownHealth { + t.Errorf("expected unknown last, got %q", changes[2].Status) + } +} + +func TestRecord_KeepsHistoryOrdered(t *testing.T) { + setupTestDB(t) + app := seedApp(t, "agent-1") + now := time.Now() + + record(t, models.ApplicationHealthChange{ApplicationId: app.Id, Status: models.Healthy, ChangedAt: now}) + record(t, models.ApplicationHealthChange{ApplicationId: app.Id, Status: models.Unhealthy, ChangedAt: now.Add(-time.Hour)}) + + changes := changesOf(t, app.Id) + if len(changes) != 2 { + t.Fatalf("expected 2 changes, got %d", len(changes)) + } + if !changes[1].ChangedAt.Equal(changes[0].ChangedAt) { + t.Errorf("expected an older change to move to the latest change time, got %v and %v", changes[0].ChangedAt, changes[1].ChangedAt) + } + latest, err := Latest(t.Context(), app.Id) + if err != nil { + t.Fatalf("Latest() error: %v", err) + } + if latest.Status != models.Unhealthy { + t.Errorf("expected the later recorded change to be the latest, got %q", latest.Status) + } +} + +func TestRecord_BoundsServices(t *testing.T) { + setupTestDB(t) + app := seedApp(t, "agent-1") + + services := make([]models.ServiceHealth, maxServices+5) + for i := range services { + services[i] = models.ServiceHealth{Name: strings.Repeat("s", maxServiceNameLength+10), Status: models.Unhealthy} + } + record(t, models.ApplicationHealthChange{ApplicationId: app.Id, Status: models.Unhealthy, Services: services}) + + stored := changesOf(t, app.Id)[0].Services + if len(stored) != maxServices { + t.Errorf("expected %d services, got %d", maxServices, len(stored)) + } + if len(stored[0].Name) != maxServiceNameLength { + t.Errorf("expected service name to be cut to %d bytes, got %d", maxServiceNameLength, len(stored[0].Name)) + } +} + +func TestSanitizeServices(t *testing.T) { + long := strings.Repeat("s", maxServiceNameLength-1) + got := sanitizeServices([]models.ServiceHealth{ + {Name: " web ", Status: models.Unhealthy}, + {Name: long + " tail", Status: models.Stopped}, + {Name: long + "ü", Status: models.Unhealthy}, + {Name: "bad\xffname", Status: models.Unhealthy}, + }) + want := []models.ServiceHealth{ + {Name: "web", Status: models.Unhealthy}, + {Name: long, Status: models.Stopped}, + {Name: long, Status: models.Unhealthy}, + {Name: "badname", Status: models.Unhealthy}, + } + if !slices.Equal(got, want) { + t.Errorf("sanitizeServices() = %q, want %q", got, want) + } + if sanitizeServices(nil) != nil { + t.Error("expected no services to stay nil") + } +} + +func TestRecordAgentOffline(t *testing.T) { + setupTestDB(t) + app := seedApp(t, "agent-1") + otherApp := seedApp(t, "agent-1") + foreignApp := seedApp(t, "agent-2") + for _, id := range []string{app.Id, otherApp.Id, foreignApp.Id} { + record(t, models.ApplicationHealthChange{ApplicationId: id, Status: models.Healthy, ChangedAt: time.Now().Add(-time.Hour)}) + } + + offlineSince := time.Now().Add(-time.Minute) + if err := RecordAgentOffline(t.Context(), "agent-1", offlineSince); err != nil { + t.Fatalf("RecordAgentOffline() error: %v", err) + } + + for _, id := range []string{app.Id, otherApp.Id} { + latest, err := Latest(t.Context(), id) + if err != nil { + t.Fatalf("Latest() error: %v", err) + } + if latest.Status != models.UnknownHealth || latest.Reason == nil || *latest.Reason != models.HealthChangeReasonAgentOffline { + t.Errorf("expected agent offline change, got %+v", latest) + } + if !latest.ChangedAt.Equal(offlineSince.UTC()) { + t.Errorf("expected change at %v, got %v", offlineSince, latest.ChangedAt) + } + } + if latest, _ := Latest(t.Context(), foreignApp.Id); latest.Status != models.Healthy { + t.Errorf("another agent's application must be untouched, got %q", latest.Status) + } +} + +func TestPrune_KeepsLatestChangeBeforeCutoff(t *testing.T) { + setupTestDB(t) + app := seedApp(t, "agent-1") + otherApp := seedApp(t, "agent-1") + now := time.Now() + + record(t, models.ApplicationHealthChange{ApplicationId: app.Id, Status: models.Healthy, ChangedAt: now.Add(-Retention - 3*time.Hour)}) + record(t, models.ApplicationHealthChange{ApplicationId: app.Id, Status: models.Unhealthy, ChangedAt: now.Add(-Retention - 2*time.Hour)}) + record(t, models.ApplicationHealthChange{ApplicationId: app.Id, Status: models.Healthy, ChangedAt: now.Add(-Retention - time.Hour)}) + record(t, models.ApplicationHealthChange{ApplicationId: app.Id, Status: models.Stopped, ChangedAt: now.Add(-time.Hour)}) + record(t, models.ApplicationHealthChange{ApplicationId: otherApp.Id, Status: models.Healthy, ChangedAt: now.Add(-Retention - time.Hour)}) + + deleted, err := Prune(t.Context(), now) + if err != nil { + t.Fatalf("Prune() error: %v", err) + } + if deleted != 2 { + t.Errorf("expected 2 deleted changes, got %d", deleted) + } + + changes := changesOf(t, app.Id) + if len(changes) != 2 || changes[0].Status != models.Healthy || changes[1].Status != models.Stopped { + t.Errorf("expected the last change before the cutoff and the recent change to remain, got %+v", changes) + } + if changes := changesOf(t, otherApp.Id); len(changes) != 1 { + t.Errorf("expected the only change of another application to remain, got %d", len(changes)) + } +} + +func TestHealthChangesAreDeletedWithApplication(t *testing.T) { + setupTestDB(t) + app := seedApp(t, "agent-1") + record(t, models.ApplicationHealthChange{ApplicationId: app.Id, Status: models.Healthy}) + + if _, err := gorm.G[models.Application](db.DB).Where("id = ?", app.Id).Delete(t.Context()); err != nil { + t.Fatalf("delete application: %v", err) + } + if changes := changesOf(t, app.Id); len(changes) != 0 { + t.Errorf("expected health changes to be deleted with the application, got %d", len(changes)) + } +} + +func TestStartPruner(t *testing.T) { + setupTestDB(t) + app := seedApp(t, "agent-1") + now := time.Now() + record(t, models.ApplicationHealthChange{ApplicationId: app.Id, Status: models.Healthy, ChangedAt: now.Add(-Retention - 2*time.Hour)}) + record(t, models.ApplicationHealthChange{ApplicationId: app.Id, Status: models.Unhealthy, ChangedAt: now.Add(-Retention - time.Hour)}) + record(t, models.ApplicationHealthChange{ApplicationId: app.Id, Status: models.Healthy, ChangedAt: now}) + + log := zerolog.Nop() + stop := StartPruner(&log) + deadline := time.Now().Add(5 * time.Second) + for len(changesOf(t, app.Id)) != 2 { + if time.Now().After(deadline) { + stop() + t.Fatalf("expected the pruner to prune on start, got %d changes", len(changesOf(t, app.Id))) + } + time.Sleep(10 * time.Millisecond) + } + + stop() + // Stopping again must not block or panic. + stop() +} diff --git a/backend/internal/hub/models/application_health_changes.go b/backend/internal/hub/models/application_health_changes.go new file mode 100644 index 00000000..e274ecae --- /dev/null +++ b/backend/internal/hub/models/application_health_changes.go @@ -0,0 +1,56 @@ +package models + +import ( + "time" + + "github.com/google/uuid" + "gorm.io/gorm" +) + +type HealthChangeReason string + +const ( + // HealthChangeReasonAgentOffline marks health that became unknown because + // the hub lost the connection to the application's agent. + HealthChangeReasonAgentOffline HealthChangeReason = "agent_offline" +) + +// ServiceHealth is the health of a single compose service at the time of a +// health change. +type ServiceHealth struct { + Name string `json:"name"` + Status HealthStatus `json:"status"` +} + +// ApplicationHealthChange records a change of an application's health. Rows +// are only written when the status differs from the previous row, so the +// status of a row holds until the next row's ChangedAt. +type ApplicationHealthChange struct { + Id string `gorm:"type:text;primaryKey"` + ApplicationId string `gorm:"type:text;not null;index:idx_application_health_changes_app_changed,priority:1"` + Application Application `gorm:"foreignKey:ApplicationId;constraint:OnUpdate:CASCADE,OnDelete:CASCADE;"` + Status HealthStatus `gorm:"type:text;not null"` + Reason *HealthChangeReason `gorm:"type:text"` + // Services holds the services that were not healthy, for incident context. + Services []ServiceHealth `gorm:"type:text;serializer:json"` + ChangedAt time.Time `gorm:"not null;index:idx_application_health_changes_app_changed,priority:2;index:idx_application_health_changes_changed"` +} + +func (ApplicationHealthChange) TableName() string { return "application_health_changes" } + +func (c *ApplicationHealthChange) BeforeCreate(tx *gorm.DB) error { + if c.Id == "" { + newId, err := uuid.NewV7() + if err != nil { + return err + } + c.Id = newId.String() + } + if c.ChangedAt.IsZero() { + c.ChangedAt = time.Now() + } + // SQLite compares timestamps as text, which only orders correctly when + // every row uses the same offset. + c.ChangedAt = c.ChangedAt.UTC() + return nil +} diff --git a/backend/internal/hub/server.go b/backend/internal/hub/server.go index acbb7c6b..0974d2da 100644 --- a/backend/internal/hub/server.go +++ b/backend/internal/hub/server.go @@ -15,6 +15,7 @@ import ( "github.com/OrcaCD/orca-cd/internal/hub/auth" "github.com/OrcaCD/orca-cd/internal/hub/crypto" "github.com/OrcaCD/orca-cd/internal/hub/db" + "github.com/OrcaCD/orca-cd/internal/hub/healthhistory" "github.com/OrcaCD/orca-cd/internal/hub/middleware" "github.com/OrcaCD/orca-cd/internal/hub/models" "github.com/OrcaCD/orca-cd/internal/hub/setupcode" @@ -158,6 +159,7 @@ func Run(cfg Config) error { defer applications.DefaultImagePullDebouncer.Stop() defer db.StartVacuumScheduler()() + defer healthhistory.StartPruner(&Log)() } router := gin.New() @@ -280,6 +282,8 @@ func recoverInterruptedState(ctx context.Context, log *zerolog.Logger) { log.Warn().Err(resetAppSyncStatus.Error).Msg("failed to reset applications stuck in syncing status") } + recoverApplicationHealth(ctx, log) + // No agent is connected yet, mark all agents as offline if _, err := gorm.G[models.Agent](db.DB). Where("status <> ?", models.AgentStatusOffline). @@ -294,3 +298,23 @@ func recoverInterruptedState(ctx context.Context, log *zerolog.Logger) { log.Info().Int64("count", recovered).Msg("marked interrupted application events as failed") } } + +// recoverApplicationHealth marks each agent offline in the health history from +// when it was last seen, as no agent is connected yet. The time the hub was +// down then shows up as a gap. The applications keep their last health so a +// change while the hub was down still notifies once the agent reconnects. +func recoverApplicationHealth(ctx context.Context, log *zerolog.Logger) { + agents, err := gorm.G[models.Agent](db.DB).Select("id", "last_seen").Find(ctx) + if err != nil { + log.Warn().Err(err).Msg("failed to load agents to recover application health") + } + for _, agent := range agents { + offlineSince := time.Now() + if agent.LastSeen != nil { + offlineSince = *agent.LastSeen + } + if err := healthhistory.RecordAgentOffline(ctx, agent.Id, offlineSince); err != nil { + log.Warn().Err(err).Str("agent_id", agent.Id).Msg("failed to record application health history for offline agent") + } + } +} diff --git a/backend/internal/hub/server_test.go b/backend/internal/hub/server_test.go index addcdeaf..b03ea3f1 100644 --- a/backend/internal/hub/server_test.go +++ b/backend/internal/hub/server_test.go @@ -291,7 +291,7 @@ func setupRecoveryTestDB(t *testing.T) { if err != nil { t.Fatalf("failed to open test db: %v", err) } - if err := testDB.AutoMigrate(&models.Agent{}, &models.Repository{}, &models.Application{}, &models.ApplicationEvent{}); err != nil { + if err := testDB.AutoMigrate(&models.Agent{}, &models.Repository{}, &models.Application{}, &models.ApplicationEvent{}, &models.ApplicationHealthChange{}); err != nil { t.Fatalf("failed to migrate: %v", err) } sqlDB, err := testDB.DB() @@ -323,10 +323,12 @@ func TestRecoverInterruptedState(t *testing.T) { if err := db.DB.Select("*").Create(&repo).Error; err != nil { t.Fatalf("failed to seed repo: %v", err) } + lastSeen := time.Now().Add(-10 * time.Minute).UTC() agent := models.Agent{ - Name: crypto.EncryptedString("test-agent"), - KeyId: crypto.EncryptedString("test-key"), - Status: models.AgentStatusOnline, + Name: crypto.EncryptedString("test-agent"), + KeyId: crypto.EncryptedString("test-key"), + Status: models.AgentStatusOnline, + LastSeen: &lastSeen, } if err := db.DB.Select("*").Create(&agent).Error; err != nil { t.Fatalf("failed to seed agent: %v", err) @@ -336,7 +338,7 @@ func TestRecoverInterruptedState(t *testing.T) { RepositoryId: repo.Id, AgentId: agent.Id, SyncStatus: models.Syncing, - HealthStatus: models.UnknownHealth, + HealthStatus: models.Healthy, Branch: "main", Path: "deploy.yml", ComposeFile: crypto.EncryptedString("services: {}\n"), @@ -344,6 +346,13 @@ func TestRecoverInterruptedState(t *testing.T) { if err := db.DB.Select("*").Create(&app).Error; err != nil { t.Fatalf("failed to seed application: %v", err) } + if err := db.DB.Create(&models.ApplicationHealthChange{ + ApplicationId: app.Id, + Status: models.Healthy, + ChangedAt: time.Now().Add(-time.Hour), + }).Error; err != nil { + t.Fatalf("failed to seed health change: %v", err) + } requestID := "interrupted-request" runningEvent := models.ApplicationEvent{ @@ -394,6 +403,21 @@ func TestRecoverInterruptedState(t *testing.T) { if recoveredApp.SyncStatus != models.OutOfSync { t.Errorf("expected app SyncStatus %q, got %q", models.OutOfSync, recoveredApp.SyncStatus) } + if recoveredApp.HealthStatus != models.Healthy { + t.Errorf("expected app HealthStatus to stay %q, got %q", models.Healthy, recoveredApp.HealthStatus) + } + + var latestHealth models.ApplicationHealthChange + if err := db.DB.Order("changed_at DESC").First(&latestHealth, "application_id = ?", app.Id).Error; err != nil { + t.Fatalf("failed to load latest health change: %v", err) + } + if latestHealth.Status != models.UnknownHealth || + latestHealth.Reason == nil || *latestHealth.Reason != models.HealthChangeReasonAgentOffline { + t.Errorf("expected agent offline health change, got %+v", latestHealth) + } + if !latestHealth.ChangedAt.Equal(lastSeen) { + t.Errorf("expected agent to be offline since it was last seen (%v), got %v", lastSeen, latestHealth.ChangedAt) + } var recoveredEvent models.ApplicationEvent if err := db.DB.First(&recoveredEvent, "id = ?", runningEvent.Id).Error; err != nil { diff --git a/backend/internal/hub/websocket/deploy_test.go b/backend/internal/hub/websocket/deploy_test.go index 4dcefb42..d2bad11e 100644 --- a/backend/internal/hub/websocket/deploy_test.go +++ b/backend/internal/hub/websocket/deploy_test.go @@ -36,7 +36,7 @@ func setupDeployTestEnv(t *testing.T) { if err != nil { t.Fatalf("failed to open test db: %v", err) } - if err := testDB.AutoMigrate(&models.Repository{}, &models.Application{}, &models.Notification{}, &models.ApplicationEvent{}); err != nil { + if err := testDB.AutoMigrate(&models.Repository{}, &models.Application{}, &models.Notification{}, &models.ApplicationEvent{}, &models.ApplicationHealthChange{}); err != nil { t.Fatalf("failed to migrate: %v", err) } diff --git a/backend/internal/hub/websocket/handler.go b/backend/internal/hub/websocket/handler.go index 38828f54..76080e76 100644 --- a/backend/internal/hub/websocket/handler.go +++ b/backend/internal/hub/websocket/handler.go @@ -12,6 +12,7 @@ import ( "github.com/OrcaCD/orca-cd/internal/hub/auth" "github.com/OrcaCD/orca-cd/internal/hub/db" + "github.com/OrcaCD/orca-cd/internal/hub/healthhistory" "github.com/OrcaCD/orca-cd/internal/hub/models" "github.com/OrcaCD/orca-cd/internal/hub/notifications" "github.com/OrcaCD/orca-cd/internal/hub/sse" @@ -177,6 +178,9 @@ func WsHandler(h *Hub, log *zerolog.Logger) gin.HandlerFunc { if applicationErr != nil { log.Error().Err(applicationErr).Str("agent_id", claims.Subject).Msg("Failed to update application status to offline") } + if historyErr := healthhistory.RecordAgentOffline(ctx, claims.Subject, time.Now()); historyErr != nil { + log.Error().Err(historyErr).Str("agent_id", claims.Subject).Msg("Failed to record application health history for offline agent") + } // Fail any deploy still in progress: the agent that owned it is gone, so // no result will arrive and the app would otherwise spin in syncing. _, syncErr := gorm.G[models.Application](db.DB). @@ -304,8 +308,11 @@ func handleApplicationStatusReport(parent context.Context, client *Client, repor if ctx.Err() != nil { return } + if status == nil { + continue + } health := reportedApplicationHealth(status) - event, err := applyReportedHealth(ctx, client.Id, status.ApplicationId, health) + event, changed, err := applyReportedHealth(ctx, client.Id, status.ApplicationId, health) if err != nil { log.Error().Err(err). Str("agent_id", client.Id). @@ -313,6 +320,12 @@ func handleApplicationStatusReport(parent context.Context, client *Client, repor Msg("failed to update application health from status report") continue } + if err := recordHealthHistory(ctx, client, status, health, changed); err != nil { + log.Error().Err(err). + Str("agent_id", client.Id). + Str("applicationId", status.ApplicationId). + Msg("failed to record application health history") + } if event != "" { notifyHealthChange(ctx, status.ApplicationId, event, health, log) } @@ -382,15 +395,16 @@ func sanitizeReportedVersion(v string) string { } // applyReportedHealth stores the reported health and returns the health event -// the change triggers, if any. Transitions are detected with conditional -// updates so concurrent reports cannot notify twice. +// the change triggers, if any, and whether the health of one of the agent's +// applications changed. Transitions are detected with conditional updates so +// concurrent reports cannot notify twice. // // Only Healthy→Unhealthy/Stopped/Degraded and back to Healthy notify; stopped // and degraded applications count as unhealthy for notifications. Unknown→Unhealthy // is deliberately silent: health is reset to Unknown on every disconnect, so it // would re-announce unhealthy apps after each agent reconnect. Apps that are // currently deploying are skipped, as the deployment reports its own outcome. -func applyReportedHealth(ctx context.Context, agentId, applicationId string, health models.HealthStatus) (models.NotificationEvent, error) { +func applyReportedHealth(ctx context.Context, agentId, applicationId string, health models.HealthStatus) (models.NotificationEvent, bool, error) { var event models.NotificationEvent transition := gorm.G[models.Application](db.DB).Where("id = ? AND agent_id = ?", applicationId, agentId) switch health { @@ -405,17 +419,45 @@ func applyReportedHealth(ctx context.Context, agentId, applicationId string, hea if event != "" { rowsAffected, err := transition.Update(ctx, "health_status", health) if err != nil { - return "", err + return "", false, err } if rowsAffected > 0 { - return event, nil + return event, true, nil } } - _, err := gorm.G[models.Application](db.DB). - Where("id = ? AND agent_id = ?", applicationId, agentId). + rowsAffected, err := gorm.G[models.Application](db.DB). + Where("id = ? AND agent_id = ? AND health_status <> ?", applicationId, agentId, health). Update(ctx, "health_status", health) - return "", err + return "", rowsAffected > 0, err +} + +// recordHealthHistory adds a reported health to the application's health +// history. Reports that do not change the application's health are only +// recorded once per connection: the history can lag behind the stored health +// after a hub restart, which records the agent as offline but keeps the health. +func recordHealthHistory(ctx context.Context, client *Client, status *messages.ApplicationStatus, health models.HealthStatus, changed bool) error { + if !changed { + if client.healthHistorySynced(status.ApplicationId) { + return nil + } + // Unchanged reports update no row, so ownership is checked separately. + owned, err := gorm.G[models.Application](db.DB). + Where("id = ? AND agent_id = ?", status.ApplicationId, client.Id). + Count(ctx, "*") + if err != nil || owned == 0 { + return err + } + } + if _, err := healthhistory.Record(ctx, models.ApplicationHealthChange{ + ApplicationId: status.ApplicationId, + Status: health, + Services: failingServices(status.Services), + }); err != nil { + return err + } + client.markHealthHistorySynced(status.ApplicationId) + return nil } func notifyHealthChange(ctx context.Context, applicationId string, event models.NotificationEvent, health models.HealthStatus, log *zerolog.Logger) { @@ -441,18 +483,31 @@ func notifyHealthChange(ctx context.Context, applicationId string, event models. // report. Agents report a partially failing application as unhealthy; when at // least one of its services is still healthy, the hub records it as degraded. func reportedApplicationHealth(status *messages.ApplicationStatus) models.HealthStatus { - health := protoHealthToModel(status.Health) + health := protoHealthToModel(status.GetHealth()) if health != models.Unhealthy { return health } for _, service := range status.Services { - if service.Health == messages.HealthStatus_HEALTH_STATUS_HEALTHY { + if service.GetHealth() == messages.HealthStatus_HEALTH_STATUS_HEALTHY { return models.Degraded } } return health } +// failingServices returns the services of a status report that are not +// healthy, which give context to incidents in the health history. +func failingServices(services []*messages.ServiceStatus) []models.ServiceHealth { + var failing []models.ServiceHealth + for _, service := range services { + if service == nil || service.GetHealth() == messages.HealthStatus_HEALTH_STATUS_HEALTHY { + continue + } + failing = append(failing, models.ServiceHealth{Name: service.GetName(), Status: protoHealthToModel(service.GetHealth())}) + } + return failing +} + func protoHealthToModel(h messages.HealthStatus) models.HealthStatus { switch h { case messages.HealthStatus_HEALTH_STATUS_HEALTHY: diff --git a/backend/internal/hub/websocket/hub.go b/backend/internal/hub/websocket/hub.go index 66fc1a0d..d61fcebb 100644 --- a/backend/internal/hub/websocket/hub.go +++ b/backend/internal/hub/websocket/hub.go @@ -27,6 +27,22 @@ type Client struct { closeOnce sync.Once closing atomic.Bool + + // healthSynced holds the applications whose health history was synced on + // this connection. Only the connection's message handler accesses it. + healthSynced map[string]struct{} +} + +func (c *Client) healthHistorySynced(applicationID string) bool { + _, synced := c.healthSynced[applicationID] + return synced +} + +func (c *Client) markHealthHistorySynced(applicationID string) { + if c.healthSynced == nil { + c.healthSynced = make(map[string]struct{}) + } + c.healthSynced[applicationID] = struct{}{} } // Close signals the WritePump to stop. diff --git a/backend/internal/hub/websocket/status_report_test.go b/backend/internal/hub/websocket/status_report_test.go index 3826a49c..ff9f7f0d 100644 --- a/backend/internal/hub/websocket/status_report_test.go +++ b/backend/internal/hub/websocket/status_report_test.go @@ -3,9 +3,11 @@ package websocket import ( "context" "testing" + "time" "github.com/OrcaCD/orca-cd/internal/hub/crypto" "github.com/OrcaCD/orca-cd/internal/hub/db" + "github.com/OrcaCD/orca-cd/internal/hub/healthhistory" "github.com/OrcaCD/orca-cd/internal/hub/models" messages "github.com/OrcaCD/orca-cd/internal/proto" "github.com/rs/zerolog" @@ -116,7 +118,7 @@ func TestApplyReportedHealth_Transitions(t *testing.T) { app := seedAppForAgent(t, "agent-1") setAppHealthAndSync(t, app.Id, tt.health, tt.sync) - event, err := applyReportedHealth(t.Context(), "agent-1", app.Id, tt.reported) + event, _, err := applyReportedHealth(t.Context(), "agent-1", app.Id, tt.reported) if err != nil { t.Fatalf("applyReportedHealth() error: %v", err) } @@ -128,7 +130,7 @@ func TestApplyReportedHealth_Transitions(t *testing.T) { } // Repeating the same report is never a transition. - event, err = applyReportedHealth(t.Context(), "agent-1", app.Id, tt.reported) + event, _, err = applyReportedHealth(t.Context(), "agent-1", app.Id, tt.reported) if err != nil { t.Fatalf("applyReportedHealth() error: %v", err) } @@ -144,7 +146,7 @@ func TestApplyReportedHealth_IgnoresOtherAgentsApps(t *testing.T) { app := seedAppForAgent(t, "agent-1") setAppHealthAndSync(t, app.Id, models.Healthy, models.Synced) - event, err := applyReportedHealth(t.Context(), "agent-2", app.Id, models.Unhealthy) + event, _, err := applyReportedHealth(t.Context(), "agent-2", app.Id, models.Unhealthy) if err != nil { t.Fatalf("applyReportedHealth() error: %v", err) } @@ -203,3 +205,88 @@ func TestReportedApplicationHealth(t *testing.T) { }) } } + +func TestHandleApplicationStatusReport_RecordsHealthHistory(t *testing.T) { + setupDeployTestEnv(t) + app := seedAppForAgent(t, "agent-1") + other := seedAppForAgent(t, "agent-2") + // Healthy→degraded sends a notification. + expectAsyncNotification(t) + + nop := zerolog.Nop() + client := &Client{Id: "agent-1"} + report := func(appID string, health messages.HealthStatus, services ...*messages.ServiceStatus) { + handleApplicationStatusReport(t.Context(), client, &messages.ApplicationStatusReport{ + Statuses: []*messages.ApplicationStatus{{ApplicationId: appID, Health: health, Services: services}}, + }, &nop) + } + + report(app.Id, messages.HealthStatus_HEALTH_STATUS_HEALTHY) + report(app.Id, messages.HealthStatus_HEALTH_STATUS_HEALTHY) + report(app.Id, messages.HealthStatus_HEALTH_STATUS_UNHEALTHY, + &messages.ServiceStatus{Name: "db", Health: messages.HealthStatus_HEALTH_STATUS_HEALTHY}, + &messages.ServiceStatus{Name: "web", Health: messages.HealthStatus_HEALTH_STATUS_UNHEALTHY}, + nil, + ) + // Spoofed: belongs to a different agent — must not be recorded. + report(other.Id, messages.HealthStatus_HEALTH_STATUS_HEALTHY) + + changes, err := gorm.G[models.ApplicationHealthChange](db.DB).Order("changed_at, id").Find(t.Context()) + if err != nil { + t.Fatalf("load health changes: %v", err) + } + if len(changes) != 2 { + t.Fatalf("expected 2 health changes, got %d: %+v", len(changes), changes) + } + if changes[0].ApplicationId != app.Id || changes[0].Status != models.Healthy { + t.Errorf("expected healthy change first, got %+v", changes[0]) + } + degraded := changes[1] + if degraded.Status != models.Degraded || len(degraded.Services) != 1 || + degraded.Services[0] != (models.ServiceHealth{Name: "web", Status: models.Unhealthy}) { + t.Errorf("expected degraded change with failing service web, got %+v", degraded) + } +} + +func TestHandleApplicationStatusReport_SyncsHealthHistoryOncePerConnection(t *testing.T) { + setupDeployTestEnv(t) + app := seedAppForAgent(t, "agent-1") + + nop := zerolog.Nop() + report := func(client *Client) { + handleApplicationStatusReport(t.Context(), client, &messages.ApplicationStatusReport{ + Statuses: []*messages.ApplicationStatus{nil, {ApplicationId: app.Id, Health: messages.HealthStatus_HEALTH_STATUS_HEALTHY}}, + }, &nop) + } + latest := func() models.HealthStatus { + t.Helper() + change, err := healthhistory.Latest(t.Context(), app.Id) + if err != nil { + t.Fatalf("Latest() error: %v", err) + } + return change.Status + } + + client := &Client{Id: "agent-1"} + report(client) + if got := latest(); got != models.Healthy { + t.Fatalf("expected healthy to be recorded, got %q", got) + } + + // A hub restart records the agent offline but keeps the stored health. + if err := healthhistory.RecordAgentOffline(t.Context(), "agent-1", time.Now()); err != nil { + t.Fatalf("RecordAgentOffline() error: %v", err) + } + report(client) + if got := latest(); got != models.UnknownHealth { + t.Errorf("an unchanged report must not be recorded again on the same connection, got %q", got) + } + + report(&Client{Id: "agent-1"}) + if got := latest(); got != models.Healthy { + t.Errorf("expected the first report of a new connection to sync the history, got %q", got) + } + if got := healthOf(t, app.Id); got != models.Healthy { + t.Errorf("expected the stored health to stay healthy, got %q", got) + } +}