diff --git a/internal/service/migrator.go b/internal/service/migrator.go index 060e1a2e..120fd41e 100644 --- a/internal/service/migrator.go +++ b/internal/service/migrator.go @@ -185,6 +185,7 @@ func MigrateUpWithConfig(db *gorm.DB, cfg *config.Config) error { &relational.AgentServiceAccountKey{}, &relational.AgentAuthEvent{}, &relational.AgentConfigRevision{}, + &relational.AgentInstance{}, &relational.UserNotificationSubscription{}, &relational.SystemNotificationDestination{}, &Heartbeat{}, @@ -1160,6 +1161,7 @@ func MigrateDown(db *gorm.DB) error { &poamrel.PoamItemMilestone{}, &poamrel.PoamItem{}, + &relational.AgentInstance{}, &relational.AgentConfigRevision{}, &relational.AgentAuthEvent{}, &relational.AgentServiceAccountKey{}, diff --git a/internal/service/relational/agent_config.go b/internal/service/relational/agent_config.go index 5cd1fd59..d88d95b3 100644 --- a/internal/service/relational/agent_config.go +++ b/internal/service/relational/agent_config.go @@ -35,3 +35,45 @@ func (*AgentConfigRevision) BeforeUpdate(*gorm.DB) error { return ErrAgentConfig // BeforeDelete keeps revisions append-only. func (*AgentConfigRevision) BeforeDelete(*gorm.DB) error { return ErrAgentConfigRevisionAppendOnly } + +// AgentInstance is one running agent process (instance id) of an agent service account, as +// last reported through a config report or an authenticated heartbeat. The displayed status +// is derived at read time (R10), not stored. +type AgentInstance struct { + UUIDModel + CreatedAt time.Time + UpdatedAt time.Time + + AgentID uuid.UUID `gorm:"type:uuid;not null;uniqueIndex:idx_agent_instance,priority:1"` + InstanceID uuid.UUID `gorm:"type:uuid;not null;uniqueIndex:idx_agent_instance,priority:2"` + CredentialID *uuid.UUID `gorm:"type:uuid"` + + Hostname *string `gorm:"type:text"` + AgentVersion *string `gorm:"type:text"` + Mode string `gorm:"type:text;not null;default:''"` // '' until the first report + Daemon *bool // R37; null until reported + + FirstSeenAt time.Time `gorm:"not null"` + LastSeenAt time.Time `gorm:"not null;index"` // authenticated heartbeats AND reports + ReportedAt *time.Time + + AppliedRevision *int64 + AttemptedRevision *int64 + ReportedStatus string `gorm:"type:text;not null;default:''"` // agent-sent; '' = never reported + ApplyReason *string `gorm:"type:text"` + ApplyError *string `gorm:"type:text"` + Truncated bool `gorm:"not null;default:false"` // R10 + Warnings datatypes.JSON `gorm:"type:jsonb"` // R41: []agentconfig.FieldError + + BaseConfig datatypes.JSON `gorm:"type:jsonb"` + EffectiveConfig datatypes.JSON `gorm:"type:jsonb"` + EffectiveDigest *string `gorm:"type:text"` + RemoteConfig datatypes.JSON `gorm:"type:jsonb"` + UnsafeChanges datatypes.JSON `gorm:"type:jsonb"` + Plugins datatypes.JSON `gorm:"type:jsonb"` // R76: []agentconfig.PluginReport + + HeartbeatConfigRevision *int64 + HeartbeatConfigDigest *string `gorm:"type:text"` +} + +func (AgentInstance) TableName() string { return "ccf_agent_instances" } diff --git a/internal/service/relational/agentcfg/derive_test.go b/internal/service/relational/agentcfg/derive_test.go new file mode 100644 index 00000000..fd09fbdd --- /dev/null +++ b/internal/service/relational/agentcfg/derive_test.go @@ -0,0 +1,173 @@ +package agentcfg_test + +import ( + "testing" + "time" + + "github.com/compliance-framework/api/internal/config" + "github.com/compliance-framework/api/internal/service/relational" + "github.com/compliance-framework/api/internal/service/relational/agentcfg" + "github.com/compliance-framework/api/pkg/agentconfig" + "github.com/stretchr/testify/assert" +) + +func i64(v int64) *int64 { return &v } + +func TestDeriveStatusAndSyncStatus(t *testing.T) { + type tc struct { + name string + instance relational.AgentInstance + desired int64 + status string + sync string + } + cases := []tc{ + { + name: "never reported", + instance: relational.AgentInstance{}, + desired: 2, + status: agentconfig.StatusUnknown, + sync: agentcfg.SyncUnknown, + }, + { + name: "heartbeat-only row in apply mode is still unknown", + instance: relational.AgentInstance{Mode: agentconfig.ModeApplySafe, HeartbeatConfigRevision: i64(1)}, + desired: 1, + status: agentconfig.StatusUnknown, + sync: agentcfg.SyncUnknown, + }, + { + name: "never reported in report mode", + instance: relational.AgentInstance{Mode: agentconfig.ModeReport}, + desired: 1, + status: agentconfig.StatusUnknown, + sync: agentcfg.SyncNotApplicable, + }, + { + name: "applied and in sync", + instance: relational.AgentInstance{Mode: agentconfig.ModeApplySafe, ReportedStatus: agentconfig.StatusApplied, AppliedRevision: i64(2)}, + desired: 2, + status: agentconfig.StatusApplied, + sync: agentcfg.SyncInSync, + }, + { + name: "no revision yet, applied nil is revision 0", + instance: relational.AgentInstance{Mode: agentconfig.ModeApplyAll, ReportedStatus: agentconfig.StatusApplied}, + desired: 0, + status: agentconfig.StatusApplied, + sync: agentcfg.SyncInSync, + }, + { + name: "pending: behind and attempted nil", + instance: relational.AgentInstance{Mode: agentconfig.ModeApplySafe, ReportedStatus: agentconfig.StatusApplied, AppliedRevision: i64(1)}, + desired: 2, + status: agentconfig.StatusPending, + sync: agentcfg.SyncOutOfSync, + }, + { + name: "pending: applied nil", + instance: relational.AgentInstance{Mode: agentconfig.ModeApplyAll, ReportedStatus: agentconfig.StatusApplied}, + desired: 1, + status: agentconfig.StatusPending, + sync: agentcfg.SyncOutOfSync, + }, + { + name: "pending: attempted older than desired", + instance: relational.AgentInstance{Mode: agentconfig.ModeApplySafe, ReportedStatus: agentconfig.StatusRejected, + AppliedRevision: i64(1), AttemptedRevision: i64(2)}, + desired: 3, + status: agentconfig.StatusPending, + sync: agentcfg.SyncOutOfSync, + }, + { + name: "rejected with attempted == desired stays rejected", + instance: relational.AgentInstance{Mode: agentconfig.ModeApplySafe, ReportedStatus: agentconfig.StatusRejected, + AppliedRevision: i64(1), AttemptedRevision: i64(2)}, + desired: 2, + status: agentconfig.StatusRejected, + sync: agentcfg.SyncOutOfSync, + }, + { + name: "failed with attempted == desired stays failed", + instance: relational.AgentInstance{Mode: agentconfig.ModeApplyAll, ReportedStatus: agentconfig.StatusFailed, + AttemptedRevision: i64(4)}, + desired: 4, + status: agentconfig.StatusFailed, + sync: agentcfg.SyncOutOfSync, + }, + { + name: "report mode returns the reported status", + instance: relational.AgentInstance{Mode: agentconfig.ModeReport, ReportedStatus: agentconfig.StatusNotApplicable, AppliedRevision: i64(0)}, + desired: 5, + status: agentconfig.StatusNotApplicable, + sync: agentcfg.SyncNotApplicable, + }, + { + name: "off mode returns the reported status", + instance: relational.AgentInstance{Mode: agentconfig.ModeOff, ReportedStatus: agentconfig.StatusNotApplicable}, + desired: 5, + status: agentconfig.StatusNotApplicable, + sync: agentcfg.SyncNotApplicable, + }, + { + name: "ahead of desired (desired revision lower) is out of sync, not pending", + instance: relational.AgentInstance{Mode: agentconfig.ModeApplySafe, ReportedStatus: agentconfig.StatusApplied, AppliedRevision: i64(3)}, + desired: 2, + status: agentconfig.StatusApplied, + sync: agentcfg.SyncOutOfSync, + }, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + assert.Equal(t, c.status, agentcfg.DeriveStatus(c.instance, c.desired)) + assert.Equal(t, c.sync, agentcfg.DeriveSyncStatus(c.instance, c.desired)) + }) + } +} + +func TestIsStale(t *testing.T) { + now := time.Date(2026, 3, 1, 12, 0, 0, 0, time.UTC) + seen := func(d time.Duration) relational.AgentInstance { + return relational.AgentInstance{LastSeenAt: now.Add(-d)} + } + + assert.False(t, agentcfg.IsStale(seen(0), now, agentcfg.Settings{})) + assert.False(t, agentcfg.IsStale(seen(10*time.Minute), now, agentcfg.Settings{}), "exactly at the default boundary is fresh") + assert.True(t, agentcfg.IsStale(seen(10*time.Minute+time.Second), now, agentcfg.Settings{})) + assert.True(t, agentcfg.IsStale(seen(2*time.Minute), now, agentcfg.Settings{InstanceStaleAfter: time.Minute})) + assert.False(t, agentcfg.IsStale(seen(30*time.Minute), now, agentcfg.Settings{InstanceStaleAfter: time.Hour})) +} + +func TestSettingsWithDefaults(t *testing.T) { + defaults := config.DefaultAgentsConfig() + d := agentcfg.Settings{MaxInstancesPerAgent: -1}.WithDefaults() + assert.Equal(t, defaults.InstanceStaleAfter, d.InstanceStaleAfter) + assert.Equal(t, defaults.InstanceRetention, d.InstanceRetention) + assert.Equal(t, defaults.OneShotInstanceRetention, d.OneShotInstanceRetention) + assert.Equal(t, defaults.MaxInstancesPerAgent, d.MaxInstancesPerAgent) + + custom := agentcfg.Settings{InstanceStaleAfter: time.Minute, InstanceRetention: time.Hour, OneShotInstanceRetention: 2 * time.Minute, MaxInstancesPerAgent: 3} + assert.Equal(t, custom, custom.WithDefaults()) +} + +func TestSettingsFromConfig(t *testing.T) { + assert.Equal(t, agentcfg.Settings{}.WithDefaults(), agentcfg.SettingsFromConfig(nil)) + assert.Equal(t, agentcfg.Settings{}.WithDefaults(), agentcfg.SettingsFromConfig(&config.Config{})) + + got := agentcfg.SettingsFromConfig(&config.Config{Agents: &config.AgentsConfig{ + InstanceStaleAfter: time.Minute, + InstanceRetention: 2 * time.Hour, + OneShotInstanceRetention: 3 * time.Minute, + MaxInstancesPerAgent: 7, + }}) + assert.Equal(t, agentcfg.Settings{ + InstanceStaleAfter: time.Minute, + InstanceRetention: 2 * time.Hour, + OneShotInstanceRetention: 3 * time.Minute, + MaxInstancesPerAgent: 7, + }, got) + + partial := agentcfg.SettingsFromConfig(&config.Config{Agents: &config.AgentsConfig{MaxInstancesPerAgent: 9}}) + assert.Equal(t, 9, partial.MaxInstancesPerAgent) + assert.Equal(t, config.DefaultAgentsConfig().InstanceRetention, partial.InstanceRetention) +} diff --git a/internal/service/relational/agentcfg/service.go b/internal/service/relational/agentcfg/service.go index 2f93baae..cb876ec9 100644 --- a/internal/service/relational/agentcfg/service.go +++ b/internal/service/relational/agentcfg/service.go @@ -7,11 +7,15 @@ import ( "encoding/json" "errors" "fmt" + "slices" + "strings" + "sync" "time" "github.com/compliance-framework/api/internal/config" "github.com/compliance-framework/api/internal/service" "github.com/compliance-framework/api/internal/service/relational" + "github.com/compliance-framework/api/pkg/agentconfig" "github.com/google/uuid" "go.uber.org/zap" "gorm.io/datatypes" @@ -19,6 +23,11 @@ import ( "gorm.io/gorm/clause" ) +// capReachedTTL is how long the service remembers that an agent is at its instance cap with +// every counted instance fresh (nothing to replace), so heartbeats from unregistered +// instances skip the locked count meanwhile. +const capReachedTTL = time.Minute + // Settings tunes instance freshness, retention and the per-agent instance cap. type Settings struct { InstanceStaleAfter time.Duration // fresh <=> last_seen_at >= now - InstanceStaleAfter @@ -67,6 +76,8 @@ var ( // ErrRevisionConflict is returned (wrapped in *RevisionConflictError) when the expected // revision is not the current one. ErrRevisionConflict = errors.New("configuration revision conflict") + // ErrInstanceLimit is returned when a new instance would exceed the per-agent cap. + ErrInstanceLimit = errors.New("instance limit reached") ) // RevisionConflictError carries the current revision of a failed CreateRevision. @@ -87,6 +98,9 @@ type Service struct { settings Settings logger *zap.SugaredLogger now func() time.Time + + capMu sync.Mutex + capReached map[uuid.UUID]time.Time // agent -> when its instance cap was last found reached } // NewService builds a Service. Zero settings take the defaults; a nil logger is a no-op. @@ -95,10 +109,11 @@ func NewService(db *gorm.DB, s Settings, logger *zap.SugaredLogger) *Service { logger = zap.NewNop().Sugar() } return &Service{ - db: db, - settings: s.WithDefaults(), - logger: logger, - now: func() time.Time { return time.Now().UTC() }, + db: db, + settings: s.WithDefaults(), + logger: logger, + now: func() time.Time { return time.Now().UTC() }, + capReached: map[uuid.UUID]time.Time{}, } } @@ -253,6 +268,279 @@ func (s *Service) CreateRevision(ctx context.Context, p CreateRevisionParams) (* return created, nil } +// ---- Instances ---- + +// reportColumns are the columns a config report overwrites. +var reportColumns = []string{ + "credential_id", "hostname", "agent_version", "mode", "daemon", + "last_seen_at", "reported_at", "applied_revision", "attempted_revision", + "reported_status", "apply_reason", "apply_error", "truncated", "warnings", + "base_config", "effective_config", "effective_digest", "remote_config", + "unsafe_changes", "plugins", "updated_at", +} + +// UpsertReport stores a (validated, re-redacted) config report. A new instance at the +// per-agent cap replaces the oldest stale instance, or fails with ErrInstanceLimit when every +// counted instance is fresh; prune-eligible rows do not count (R37). +func (s *Service) UpsertReport(ctx context.Context, agentID uuid.UUID, credentialID *uuid.UUID, instanceID uuid.UUID, r agentconfig.Report) error { + now := s.now() + row, err := reportRow(agentID, credentialID, instanceID, r, now) + if err != nil { + return err + } + return s.upsert(ctx, row, reportColumns, false) +} + +func reportRow(agentID uuid.UUID, credentialID *uuid.UUID, instanceID uuid.UUID, r agentconfig.Report, now time.Time) (*relational.AgentInstance, error) { + daemon := r.Daemon + row := &relational.AgentInstance{ + AgentID: agentID, + InstanceID: instanceID, + CredentialID: credentialID, + Hostname: optionalString(r.Hostname), + AgentVersion: optionalString(r.AgentVersion), + Mode: r.Mode, + Daemon: &daemon, + FirstSeenAt: now, + LastSeenAt: now, + ReportedAt: &now, + AppliedRevision: r.AppliedRevision, + AttemptedRevision: r.AttemptedRevision, + ReportedStatus: r.Status, + ApplyReason: optionalString(r.Reason), + ApplyError: r.Error, + Truncated: r.Truncated, + BaseConfig: rawOrNil(r.Base), + EffectiveConfig: rawOrNil(r.Effective), + EffectiveDigest: optionalString(r.EffectiveDigest), + CreatedAt: now, + UpdatedAt: now, + } + var err error + if row.Warnings, err = jsonOrNil(r.Warnings); err != nil { + return nil, err + } + if r.RemoteConfig != nil { + if row.RemoteConfig, err = jsonOrNil(r.RemoteConfig); err != nil { + return nil, err + } + } + if row.UnsafeChanges, err = jsonOrNil(r.Unsafe); err != nil { + return nil, err + } + if row.Plugins, err = jsonOrNil(r.Plugins); err != nil { + return nil, err + } + return row, nil +} + +// TouchFromHeartbeat records an authenticated heartbeat (R11). With a digest it upserts +// only last_seen_at, credential_id and the heartbeat_* columns (a new row keeps an empty +// reported status; over the cap it is silently skipped, with a warning at most once per +// agent per capReachedTTL, and for capReachedTTL after the cap was found reached a new +// instance skips the cap check altogether). Without a digest (an old agent, or mode off) it +// only updates last_seen_at of an existing row and never inserts, so old agents' per-reload +// random instance ids never flood the table. +func (s *Service) TouchFromHeartbeat(ctx context.Context, agentID uuid.UUID, credentialID *uuid.UUID, instanceID uuid.UUID, rev *int64, digest *string) error { + now := s.now() + if digest == nil { + return s.db.WithContext(ctx).Model(&relational.AgentInstance{}). + Where("agent_id = ? AND instance_id = ?", agentID, instanceID). + Updates(map[string]any{"last_seen_at": now, "updated_at": now}).Error + } + row := &relational.AgentInstance{ + AgentID: agentID, + InstanceID: instanceID, + CredentialID: credentialID, + FirstSeenAt: now, + LastSeenAt: now, + HeartbeatConfigRevision: rev, + HeartbeatConfigDigest: digest, + CreatedAt: now, + UpdatedAt: now, + } + err := s.upsert(ctx, row, []string{"last_seen_at", "credential_id", "heartbeat_config_revision", "heartbeat_config_digest", "updated_at"}, true) + if errors.Is(err, ErrInstanceLimit) { + return nil + } + return err +} + +// upsert updates the instance row when it exists, else inserts it after the cap check under +// a lock on the agent row (so concurrent first reports cannot overshoot the cap). The update +// writes exactly the given columns from row, zero values included; UpdateColumns (no hooks) +// keeps updated_at at row.UpdatedAt (the service clock). With skipWhenCapped, a new instance +// of an agent whose cap was found reached within capReachedTTL fails with ErrInstanceLimit +// without the locked count. +func (s *Service) upsert(ctx context.Context, row *relational.AgentInstance, columns []string, skipWhenCapped bool) error { + db := s.db.WithContext(ctx) + res := db.Model(&relational.AgentInstance{}). + Where("agent_id = ? AND instance_id = ?", row.AgentID, row.InstanceID). + Select(columns). + UpdateColumns(row) + if res.Error != nil { + return res.Error + } + if res.RowsAffected > 0 { + return nil + } + if skipWhenCapped && s.capReachedRecently(row.AgentID) { + return ErrInstanceLimit + } + err := s.insertUnderCap(db, row, columns) + switch { + case errors.Is(err, ErrInstanceLimit): + if s.markCapReached(row.AgentID) { + s.logger.Warnw("Agent instance cap reached; new instances are not recorded", + "agentID", row.AgentID, "instanceID", row.InstanceID, "max", s.settings.MaxInstancesPerAgent) + } + case err == nil: + s.clearCapReached(row.AgentID) + } + return err +} + +// insertUnderCap inserts a new instance row under the per-agent cap (R37). When the agent +// already has MaxInstancesPerAgent non-prunable instances, the oldest stale one among them is +// deleted to make room (instance ids change on restart, so a restarted daemon's old ids must +// not lock out its new ones); when every one of them is fresh it fails with +// ErrInstanceLimit. So the non-prunable rows of an agent never exceed the cap. +func (s *Service) insertUnderCap(db *gorm.DB, row *relational.AgentInstance, columns []string) error { + return db.Transaction(func(tx *gorm.DB) error { + var agent relational.Agent + if err := tx.Clauses(clause.Locking{Strength: clause.LockingStrengthUpdate}). + Select("id").Where("id = ?", row.AgentID).Take(&agent).Error; err != nil { + if errors.Is(err, gorm.ErrRecordNotFound) { + return ErrNotFound + } + return err + } + now := row.LastSeenAt + count, err := s.countActiveInstances(tx, row.AgentID, now) + if err != nil { + return err + } + if count >= int64(s.settings.MaxInstancesPerAgent) { + evicted, err := s.evictOldestStaleInstance(tx, row.AgentID, now) + if err != nil { + return err + } + if !evicted { + return ErrInstanceLimit + } + } + return tx.Clauses(clause.OnConflict{ + Columns: []clause.Column{{Name: "agent_id"}, {Name: "instance_id"}}, + DoUpdates: clause.AssignmentColumns(columns), + }).Create(row).Error + }) +} + +// notPrunableClause selects the instances that are not eligible for pruning (R37); its two +// arguments are the one-shot and the daemon retention cutoffs. "daemon IS FALSE" (not +// "daemon = false") keeps unknown-daemon rows counted like daemon rows, as PruneInstances +// keeps them: with "= false" a NULL daemon makes the NOT(...) NULL and the row silently +// drops out. +const notPrunableClause = "NOT ((daemon IS FALSE AND last_seen_at < ?) OR last_seen_at < ?)" + +// countActiveInstances counts the instances that are not eligible for pruning (R37). +func (s *Service) countActiveInstances(tx *gorm.DB, agentID uuid.UUID, now time.Time) (int64, error) { + var count int64 + err := tx.Model(&relational.AgentInstance{}). + Where("agent_id = ?", agentID). + Where(notPrunableClause, now.Add(-s.settings.OneShotInstanceRetention), now.Add(-s.settings.InstanceRetention)). + Count(&count).Error + return count, err +} + +// evictOldestStaleInstance deletes the least recently seen stale instance (IsStale: +// last_seen_at < now - InstanceStaleAfter) among the agent's non-prunable ones, the rows the +// cap counts, and reports whether there was one. +func (s *Service) evictOldestStaleInstance(tx *gorm.DB, agentID uuid.UUID, now time.Time) (bool, error) { + var victim relational.AgentInstance + res := tx.Select("id", "instance_id"). + Where("agent_id = ? AND last_seen_at < ?", agentID, now.Add(-s.settings.InstanceStaleAfter)). + Where(notPrunableClause, now.Add(-s.settings.OneShotInstanceRetention), now.Add(-s.settings.InstanceRetention)). + Order("last_seen_at ASC, instance_id"). + Limit(1). + Find(&victim) + if res.Error != nil || res.RowsAffected == 0 { + return false, res.Error + } + if err := tx.Delete(&relational.AgentInstance{}, "id = ?", victim.ID).Error; err != nil { + return false, err + } + s.logger.Debugw("Agent instance cap reached; replaced the oldest stale instance", + "agentID", agentID, "evictedInstanceID", victim.InstanceID) + return true, nil +} + +// capReachedRecently reports whether agentID's instance cap was found reached within +// capReachedTTL. +func (s *Service) capReachedRecently(agentID uuid.UUID) bool { + s.capMu.Lock() + defer s.capMu.Unlock() + at, ok := s.capReached[agentID] + return ok && s.now().Sub(at) < capReachedTTL +} + +// markCapReached records that agentID's instance cap is reached and reports whether that is +// news (not already recorded within capReachedTTL), i.e. whether to log it. +func (s *Service) markCapReached(agentID uuid.UUID) bool { + s.capMu.Lock() + defer s.capMu.Unlock() + now := s.now() + if at, ok := s.capReached[agentID]; ok && now.Sub(at) < capReachedTTL { + return false + } + s.capReached[agentID] = now + return true +} + +// clearCapReached forgets that agentID's cap was reached (a new instance was inserted). +func (s *Service) clearCapReached(agentID uuid.UUID) { + s.capMu.Lock() + defer s.capMu.Unlock() + delete(s.capReached, agentID) +} + +// summaryColumns are the instance columns loaded for list views (no base/effective). +var summaryColumns = []string{ + "id", "created_at", "updated_at", "agent_id", "instance_id", "credential_id", + "hostname", "agent_version", "mode", "daemon", "first_seen_at", "last_seen_at", + "reported_at", "applied_revision", "attempted_revision", "reported_status", + "apply_reason", "apply_error", "truncated", "warnings", "effective_digest", + "remote_config", "unsafe_changes", "plugins", + "heartbeat_config_revision", "heartbeat_config_digest", +} + +// ListInstances returns an agent's instances (most recently seen first) without the heavy +// base/effective columns. It is deliberately unpaginated: the UI needs every row for its +// counts, the instance count is capped and pruned, and normalizeReport bounds the summary +// columns (warnings, unsafe changes, plugins). +func (s *Service) ListInstances(ctx context.Context, agentID uuid.UUID) ([]relational.AgentInstance, error) { + var out []relational.AgentInstance + err := s.db.WithContext(ctx). + Select(summaryColumns). + Where("agent_id = ?", agentID). + Order("last_seen_at DESC, instance_id"). + Find(&out).Error + return out, err +} + +// GetInstance returns one instance of an agent (all columns) or ErrNotFound. +func (s *Service) GetInstance(ctx context.Context, agentID, instanceID uuid.UUID) (*relational.AgentInstance, error) { + var out relational.AgentInstance + err := s.db.WithContext(ctx).Where("agent_id = ? AND instance_id = ?", agentID, instanceID).Take(&out).Error + if errors.Is(err, gorm.ErrRecordNotFound) { + return nil, ErrNotFound + } + if err != nil { + return nil, err + } + return &out, nil +} + // DeleteRevisionsForAgent removes every configuration revision of an agent (agent deletion). // It is the purge path for an overlay that held a secret, so it bypasses the append-only // BeforeDelete hook, which still blocks every other delete of a revision. @@ -261,3 +549,85 @@ func DeleteRevisionsForAgent(tx *gorm.DB, agentID uuid.UUID) error { Where("agent_id = ?", agentID). Delete(&relational.AgentConfigRevision{}).Error } + +// ---- Derived state ---- + +// IsStale reports whether an instance was last seen before now - InstanceStaleAfter. +func IsStale(i relational.AgentInstance, now time.Time, s Settings) bool { + s = s.WithDefaults() + return i.LastSeenAt.Before(now.Add(-s.InstanceStaleAfter)) +} + +// DeriveStatus returns the displayed status of an instance (R10): unknown when it never +// reported; the reported status in report/off mode; pending when an apply-mode instance is +// behind the desired revision and has not attempted it yet; otherwise the reported status. +func DeriveStatus(i relational.AgentInstance, desired int64) string { + switch { + case i.ReportedStatus == "": + return agentconfig.StatusUnknown + case i.Mode == agentconfig.ModeReport || i.Mode == agentconfig.ModeOff: + return i.ReportedStatus + case desired > deref(i.AppliedRevision) && + (i.AttemptedRevision == nil || *i.AttemptedRevision < desired): + return agentconfig.StatusPending + default: + return i.ReportedStatus + } +} + +// Sync statuses. +const ( + SyncInSync = "in-sync" + SyncOutOfSync = "out-of-sync" + SyncNotApplicable = "not-applicable" + SyncUnknown = "unknown" +) + +// DeriveSyncStatus: report/off => not-applicable; never reported => unknown; +// COALESCE(applied,0) == desired => in-sync; else out-of-sync. +func DeriveSyncStatus(i relational.AgentInstance, desired int64) string { + switch { + case i.Mode == agentconfig.ModeReport || i.Mode == agentconfig.ModeOff: + return SyncNotApplicable + case i.ReportedStatus == "": + return SyncUnknown + case deref(i.AppliedRevision) == desired: + return SyncInSync + default: + return SyncOutOfSync + } +} + +func deref(v *int64) int64 { + if v == nil { + return 0 + } + return *v +} + +func optionalString(s string) *string { + s = strings.TrimSpace(s) + if s == "" { + return nil + } + return &s +} + +func rawOrNil(raw json.RawMessage) datatypes.JSON { + if len(raw) == 0 || string(raw) == "null" { + return nil + } + return datatypes.JSON(raw) +} + +// jsonOrNil encodes v, storing SQL NULL for nil/empty slices. +func jsonOrNil[T any](v T) (datatypes.JSON, error) { + raw, err := json.Marshal(v) + if err != nil { + return nil, err + } + if slices.Contains([]string{"null", "[]"}, string(raw)) { + return nil, nil + } + return datatypes.JSON(raw), nil +} diff --git a/internal/service/relational/agentcfg/service_integration_test.go b/internal/service/relational/agentcfg/service_integration_test.go index 171a89e4..8aa41c9c 100644 --- a/internal/service/relational/agentcfg/service_integration_test.go +++ b/internal/service/relational/agentcfg/service_integration_test.go @@ -14,6 +14,7 @@ import ( "github.com/compliance-framework/api/internal/service/relational" "github.com/compliance-framework/api/internal/service/relational/agentcfg" "github.com/compliance-framework/api/internal/tests" + "github.com/compliance-framework/api/pkg/agentconfig" "github.com/google/uuid" "github.com/stretchr/testify/suite" "gorm.io/datatypes" @@ -62,8 +63,71 @@ func (s *AgentCfgServiceIntegrationSuite) createRevision(agentID uuid.UUID, expe return rev } +// insertInstance writes an instance row directly (bypassing the cap), seen at lastSeen. +func (s *AgentCfgServiceIntegrationSuite) insertInstance(agentID uuid.UUID, daemon *bool, lastSeen time.Time) uuid.UUID { + instanceID := uuid.New() + row := &relational.AgentInstance{ + AgentID: agentID, + InstanceID: instanceID, + Mode: agentconfig.ModeApplySafe, + Daemon: daemon, + FirstSeenAt: lastSeen, + LastSeenAt: lastSeen, + CreatedAt: lastSeen, + UpdatedAt: lastSeen, + } + s.Require().NoError(s.DB.Create(row).Error) + return instanceID +} + +// deleteOneInstance deletes one (any) instance row of an agent. +func (s *AgentCfgServiceIntegrationSuite) deleteOneInstance(agentID uuid.UUID) { + var victim relational.AgentInstance + s.Require().NoError(s.DB.Where("agent_id = ?", agentID).First(&victim).Error) + s.Require().NoError(s.DB.Delete(&victim).Error) +} + +func (s *AgentCfgServiceIntegrationSuite) countInstances(agentID uuid.UUID) int64 { + var n int64 + s.Require().NoError(s.DB.Model(&relational.AgentInstance{}).Where("agent_id = ?", agentID).Count(&n).Error) + return n +} + +// reportAt stores a report with the service clock set to at, then restores the clock. +func (s *AgentCfgServiceIntegrationSuite) reportAt(svc *agentcfg.Service, at time.Time, agentID, instanceID uuid.UUID, r agentconfig.Report) error { + saved := s.now + s.now = at + defer func() { s.now = saved }() + return svc.UpsertReport(s.ctx, agentID, nil, instanceID, r) +} + +func applyReport(mode string, base string) agentconfig.Report { + r := agentconfig.Report{ + Mode: mode, + Daemon: true, + AppliedRevision: ptr(int64(1)), + Status: agentconfig.StatusApplied, + EffectiveDigest: "sha256:abc", + } + if base != "" { + r.Base = json.RawMessage(base) + r.Effective = json.RawMessage(base) + } + return r +} + func ptr[T any](v T) *T { return &v } +func mustJSON(v any) string { + raw, err := json.Marshal(v) + if err != nil { + panic(err) + } + return string(raw) +} + +const baseConfig = `{"daemon":true,"verbosity":0,"api":{"url":"http://api:8080","auth":{"client_id":"cid"}},"plugins":{"p1":{"source":"ghcr.io/x/p1:v1","policies":["ghcr.io/x/pol:v1"]}}}` + // ---- Revisions ---- func (s *AgentCfgServiceIntegrationSuite) TestRevisionHooksAreAppendOnly() { @@ -258,6 +322,332 @@ func (s *AgentCfgServiceIntegrationSuite) TestListRevisionsPagesNewestFirst() { // ---- Instances ---- +func (s *AgentCfgServiceIntegrationSuite) TestUpsertReportInsertsThenUpdates() { + agentID := s.newAgent("report") + instanceID := uuid.New() + credID := uuid.New() + t0 := s.now + + warnings := []agentconfig.FieldError{{Path: "/plugins/p1/foo", Code: "unknown-field", Message: "unknown"}} + plugins := []agentconfig.PluginReport{{Name: "p1", Source: "ghcr.io/x/p1:v1", LibVersion: "v0.7.1"}} + unsafe := []agentconfig.Change{{Path: "/plugins/p1/source", Safety: agentconfig.Unsafe, Reason: "source-changed"}} + remote := &agentconfig.RemoteConfig{Mode: agentconfig.ModeApplyAll, PollInterval: "30s", TrustedSources: []string{"ghcr.io/x/*"}} + effective := `{"daemon":true,"verbosity":1,"plugins":{}}` + + err := s.svc.UpsertReport(s.ctx, agentID, &credID, instanceID, agentconfig.Report{ + Hostname: " host-1 ", AgentVersion: "v1.2.3", Mode: agentconfig.ModeApplyAll, Daemon: true, + AppliedRevision: ptr(int64(1)), AttemptedRevision: ptr(int64(2)), Status: agentconfig.StatusFailed, + Reason: agentconfig.ReasonDownloadFailed, Error: ptr("oci pull failed"), Truncated: true, + Warnings: warnings, Base: json.RawMessage(baseConfig), Effective: json.RawMessage(effective), + EffectiveDigest: "sha256:eff", Unsafe: unsafe, + RemoteConfig: remote, Plugins: plugins, + }) + s.Require().NoError(err) + + got, err := s.svc.GetInstance(s.ctx, agentID, instanceID) + s.Require().NoError(err) + firstID := *got.ID + s.Equal(credID, *got.CredentialID) + s.Equal("host-1", *got.Hostname) + s.Equal("v1.2.3", *got.AgentVersion) + s.Equal(agentconfig.ModeApplyAll, got.Mode) + s.Require().NotNil(got.Daemon) + s.True(*got.Daemon) + s.Equal(int64(1), *got.AppliedRevision) + s.Equal(int64(2), *got.AttemptedRevision) + s.Equal(agentconfig.StatusFailed, got.ReportedStatus) + s.Equal(agentconfig.ReasonDownloadFailed, *got.ApplyReason) + s.Equal("oci pull failed", *got.ApplyError) + s.True(got.Truncated) + s.JSONEq(mustJSON(warnings), string(got.Warnings)) + s.JSONEq(baseConfig, string(got.BaseConfig)) + s.JSONEq(effective, string(got.EffectiveConfig)) + s.Equal("sha256:eff", *got.EffectiveDigest) + s.JSONEq(mustJSON(remote), string(got.RemoteConfig)) + s.JSONEq(mustJSON(unsafe), string(got.UnsafeChanges)) + s.JSONEq(mustJSON(plugins), string(got.Plugins)) + s.True(got.FirstSeenAt.Equal(t0)) + s.True(got.LastSeenAt.Equal(t0)) + s.Require().NotNil(got.ReportedAt) + s.True(got.ReportedAt.Equal(t0)) + s.Nil(got.HeartbeatConfigDigest) + + // Second report from the same instance updates the same row. + t1 := t0.Add(5 * time.Minute) + s.now = t1 + err = s.svc.UpsertReport(s.ctx, agentID, &credID, instanceID, agentconfig.Report{ + Mode: agentconfig.ModeApplySafe, Daemon: false, AppliedRevision: ptr(int64(2)), + Status: agentconfig.StatusApplied, Base: json.RawMessage(baseConfig), EffectiveDigest: "sha256:eff2", + }) + s.Require().NoError(err) + s.Equal(int64(1), s.countInstances(agentID)) + + got, err = s.svc.GetInstance(s.ctx, agentID, instanceID) + s.Require().NoError(err) + s.Equal(firstID, *got.ID) + s.True(got.FirstSeenAt.Equal(t0), "first_seen_at is kept") + s.True(got.LastSeenAt.Equal(t1)) + s.True(got.ReportedAt.Equal(t1)) + s.Equal(agentconfig.ModeApplySafe, got.Mode) + s.False(*got.Daemon) + s.Equal(int64(2), *got.AppliedRevision) + s.Nil(got.AttemptedRevision) + s.Equal(agentconfig.StatusApplied, got.ReportedStatus) + s.Nil(got.ApplyReason) + s.Nil(got.ApplyError) + s.Nil(got.Hostname) + s.False(got.Truncated) + s.Nil(got.Warnings) + s.Nil(got.EffectiveConfig) + s.Nil(got.RemoteConfig) + s.Nil(got.UnsafeChanges) + s.Nil(got.Plugins) + s.Equal("sha256:eff2", *got.EffectiveDigest) +} + +func (s *AgentCfgServiceIntegrationSuite) TestInstanceCapCountsOnlyNonPrunableRows() { + svc := s.newService(agentcfg.Settings{MaxInstancesPerAgent: 2}) + report := applyReport(agentconfig.ModeApplySafe, baseConfig) + + agentA := s.newAgent("cap-a") + i1, i2 := uuid.New(), uuid.New() + s.Require().NoError(svc.UpsertReport(s.ctx, agentA, nil, i1, report)) + s.Require().NoError(svc.UpsertReport(s.ctx, agentA, nil, i2, report)) + err := svc.UpsertReport(s.ctx, agentA, nil, uuid.New(), report) + s.ErrorIs(err, agentcfg.ErrInstanceLimit, "every counted instance is fresh: nothing to replace") + s.Equal(int64(2), s.countInstances(agentA)) + s.NoError(svc.UpsertReport(s.ctx, agentA, nil, i1, report), "an existing instance can still report") + s.NoError(svc.UpsertReport(s.ctx, s.newAgent("cap-other"), nil, uuid.New(), report), "the cap is per agent") + + // Prune-eligible rows do not count toward the cap (and are never evicted). + agentB := s.newAgent("cap-b") + s.insertInstance(agentB, ptr(false), s.now.Add(-25*time.Hour)) // one-shot, older than 24h + s.insertInstance(agentB, ptr(true), s.now.Add(-721*time.Hour)) // daemon, older than 720h + s.insertInstance(agentB, nil, s.now.Add(-800*time.Hour)) // unknown, older than 720h + s.insertInstance(agentB, ptr(false), s.now.Add(-721*time.Hour)) // both + s.Require().NoError(svc.UpsertReport(s.ctx, agentB, nil, uuid.New(), report)) + s.Require().NoError(svc.UpsertReport(s.ctx, agentB, nil, uuid.New(), report)) + s.ErrorIs(svc.UpsertReport(s.ctx, agentB, nil, uuid.New(), report), agentcfg.ErrInstanceLimit) + s.Equal(int64(6), s.countInstances(agentB)) + + // Non-prunable fresh rows count: a one-shot or an unknown (daemon NULL) instance seen + // within InstanceStaleAfter (10m by default). + agentC := s.newAgent("cap-c") + s.insertInstance(agentC, ptr(false), s.now.Add(-time.Minute)) + s.insertInstance(agentC, nil, s.now.Add(-10*time.Minute)) // exactly at the threshold: fresh + s.ErrorIs(svc.UpsertReport(s.ctx, agentC, nil, uuid.New(), report), agentcfg.ErrInstanceLimit) + + s.ErrorIs(svc.UpsertReport(s.ctx, uuid.New(), nil, uuid.New(), report), agentcfg.ErrNotFound) +} + +// At the cap, a new instance replaces the oldest stale counted instance (e.g. a restarted +// DaemonSet's old instance ids), so the counted rows never exceed the cap. +func (s *AgentCfgServiceIntegrationSuite) TestInstanceCapReplacesOldestStaleInstance() { + svc := s.newService(agentcfg.Settings{MaxInstancesPerAgent: 3}) + report := applyReport(agentconfig.ModeApplySafe, baseConfig) + agentID := s.newAgent("cap-evict") + instanceExists := func(id uuid.UUID) bool { + _, err := svc.GetInstance(s.ctx, agentID, id) + if errors.Is(err, agentcfg.ErrNotFound) { + return false + } + s.Require().NoError(err) + return true + } + + fresh := s.insertInstance(agentID, ptr(true), s.now.Add(-time.Minute)) + older := s.insertInstance(agentID, ptr(true), s.now.Add(-2*time.Hour)) + oldest := s.insertInstance(agentID, nil, s.now.Add(-100*time.Hour)) + prunable := s.insertInstance(agentID, ptr(false), s.now.Add(-48*time.Hour)) // not counted, never evicted + + // A report from a new instance replaces the oldest stale counted row. + r1 := uuid.New() + s.Require().NoError(svc.UpsertReport(s.ctx, agentID, nil, r1, report)) + s.True(instanceExists(r1)) + s.False(instanceExists(oldest), "the oldest stale instance is replaced") + s.True(instanceExists(older)) + s.True(instanceExists(prunable), "prune-eligible rows are left to the prune job") + s.Equal(int64(4), s.countInstances(agentID)) + + // A heartbeat from a new instance does the same. + h1 := uuid.New() + s.Require().NoError(svc.TouchFromHeartbeat(s.ctx, agentID, nil, h1, nil, ptr("sha256:hb"))) + s.True(instanceExists(h1)) + s.False(instanceExists(older)) + s.True(instanceExists(fresh)) + s.Equal(int64(4), s.countInstances(agentID)) + + // Now every counted instance is fresh: reports get ErrInstanceLimit, heartbeats are + // skipped, and nothing is deleted. + s.ErrorIs(svc.UpsertReport(s.ctx, agentID, nil, uuid.New(), report), agentcfg.ErrInstanceLimit) + s.NoError(svc.TouchFromHeartbeat(s.ctx, agentID, nil, uuid.New(), nil, ptr("sha256:hb"))) + s.Equal(int64(4), s.countInstances(agentID)) + s.True(instanceExists(fresh)) + s.True(instanceExists(r1)) + s.True(instanceExists(h1)) + + // Many new instances in a row never grow the counted rows past the cap. + s.now = s.now.Add(time.Hour) // every row is stale now + for i := 0; i < 10; i++ { + s.Require().NoError(svc.UpsertReport(s.ctx, agentID, nil, uuid.New(), report)) + s.now = s.now.Add(11 * time.Minute) + } + var counted int64 + s.Require().NoError(s.DB.Model(&relational.AgentInstance{}). + Where("agent_id = ? AND (daemon IS NULL OR daemon = true)", agentID).Count(&counted).Error) + s.Equal(int64(3), counted) +} + +// Once the cap is found reached, heartbeats from new instances skip the insert path (no +// locked count) for capReachedTTL (1m); reports always check the cap. +func (s *AgentCfgServiceIntegrationSuite) TestHeartbeatCapReachedIsCached() { + svc := s.newService(agentcfg.Settings{MaxInstancesPerAgent: 2}) + agentID := s.newAgent("cap-cache") + digest := ptr("sha256:hb") + s.insertInstance(agentID, ptr(true), s.now) + freed := s.insertInstance(agentID, ptr(true), s.now) + + s.NoError(svc.TouchFromHeartbeat(s.ctx, agentID, nil, uuid.New(), nil, digest)) + s.Equal(int64(2), s.countInstances(agentID)) + + // A slot frees up, but within the TTL a new heartbeating instance is still skipped. + s.Require().NoError(s.DB.Where("agent_id = ? AND instance_id = ?", agentID, freed).Delete(&relational.AgentInstance{}).Error) + s.now = s.now.Add(30 * time.Second) + s.NoError(svc.TouchFromHeartbeat(s.ctx, agentID, nil, uuid.New(), nil, digest)) + s.Equal(int64(1), s.countInstances(agentID)) + + // Other agents are not affected. + other := s.newAgent("cap-cache-other") + s.NoError(svc.TouchFromHeartbeat(s.ctx, other, nil, uuid.New(), nil, digest)) + s.Equal(int64(1), s.countInstances(other)) + + // After the TTL the cap is checked again and the instance is recorded. + s.now = s.now.Add(31 * time.Second) + s.NoError(svc.TouchFromHeartbeat(s.ctx, agentID, nil, uuid.New(), nil, digest)) + s.Equal(int64(2), s.countInstances(agentID)) + + // Reports ignore the cache: they find the cap reached (409)... + report := applyReport(agentconfig.ModeApplySafe, baseConfig) + s.ErrorIs(svc.UpsertReport(s.ctx, agentID, nil, uuid.New(), report), agentcfg.ErrInstanceLimit) + // ...and are recorded as soon as a slot is free, which also clears the cache. + s.deleteOneInstance(agentID) + s.Equal(int64(1), s.countInstances(agentID)) + s.Require().NoError(svc.UpsertReport(s.ctx, agentID, nil, uuid.New(), report)) + s.Equal(int64(2), s.countInstances(agentID)) + s.deleteOneInstance(agentID) + s.NoError(svc.TouchFromHeartbeat(s.ctx, agentID, nil, uuid.New(), nil, digest)) + s.Equal(int64(2), s.countInstances(agentID), "a successful insert clears the cached cap") +} + +func (s *AgentCfgServiceIntegrationSuite) TestTouchFromHeartbeat() { + agentID := s.newAgent("heartbeat") + credID := uuid.New() + + // Without a digest nothing is ever inserted. + s.Require().NoError(s.svc.TouchFromHeartbeat(s.ctx, agentID, &credID, uuid.New(), ptr(int64(3)), nil)) + s.Equal(int64(0), s.countInstances(agentID)) + + // Without a digest an existing row only gets last_seen_at. + reported := uuid.New() + t0 := s.now + s.Require().NoError(s.svc.UpsertReport(s.ctx, agentID, nil, reported, applyReport(agentconfig.ModeApplySafe, baseConfig))) + t1 := t0.Add(3 * time.Minute) + s.now = t1 + s.Require().NoError(s.svc.TouchFromHeartbeat(s.ctx, agentID, &credID, reported, ptr(int64(3)), nil)) + got, err := s.svc.GetInstance(s.ctx, agentID, reported) + s.Require().NoError(err) + s.True(got.LastSeenAt.Equal(t1)) + s.True(got.ReportedAt.Equal(t0)) + s.Nil(got.CredentialID) + s.Nil(got.HeartbeatConfigRevision) + s.Nil(got.HeartbeatConfigDigest) + s.Equal(agentconfig.StatusApplied, got.ReportedStatus) + + // With a digest a new row is inserted with an empty reported status. + hbOnly := uuid.New() + s.Require().NoError(s.svc.TouchFromHeartbeat(s.ctx, agentID, &credID, hbOnly, ptr(int64(3)), ptr("sha256:hb"))) + got, err = s.svc.GetInstance(s.ctx, agentID, hbOnly) + s.Require().NoError(err) + s.Equal("", got.ReportedStatus) + s.Equal("", got.Mode) + s.Nil(got.Daemon) + s.Nil(got.ReportedAt) + s.Nil(got.BaseConfig) + s.Equal(credID, *got.CredentialID) + s.Equal(int64(3), *got.HeartbeatConfigRevision) + s.Equal("sha256:hb", *got.HeartbeatConfigDigest) + s.True(got.FirstSeenAt.Equal(t1)) + s.True(got.LastSeenAt.Equal(t1)) + + // With a digest an existing reported row keeps its report columns. + t2 := t1.Add(time.Minute) + s.now = t2 + s.Require().NoError(s.svc.TouchFromHeartbeat(s.ctx, agentID, &credID, reported, ptr(int64(4)), ptr("sha256:hb2"))) + got, err = s.svc.GetInstance(s.ctx, agentID, reported) + s.Require().NoError(err) + s.True(got.LastSeenAt.Equal(t2)) + s.Equal(int64(4), *got.HeartbeatConfigRevision) + s.Equal("sha256:hb2", *got.HeartbeatConfigDigest) + s.Equal(credID, *got.CredentialID) + s.Equal(agentconfig.StatusApplied, got.ReportedStatus) + s.Equal(agentconfig.ModeApplySafe, got.Mode) + s.JSONEq(baseConfig, string(got.BaseConfig)) + s.True(got.ReportedAt.Equal(t0)) + s.Equal(int64(2), s.countInstances(agentID)) + + // Over the cap a heartbeat with a digest is silently skipped. + capped := s.newService(agentcfg.Settings{MaxInstancesPerAgent: 2}) + s.NoError(capped.TouchFromHeartbeat(s.ctx, agentID, &credID, uuid.New(), ptr(int64(4)), ptr("sha256:hb3"))) + s.Equal(int64(2), s.countInstances(agentID)) + s.NoError(capped.TouchFromHeartbeat(s.ctx, agentID, &credID, hbOnly, ptr(int64(5)), ptr("sha256:hb4")), + "existing instances still record heartbeats over the cap") + got, err = s.svc.GetInstance(s.ctx, agentID, hbOnly) + s.Require().NoError(err) + s.Equal(int64(5), *got.HeartbeatConfigRevision) +} + +func (s *AgentCfgServiceIntegrationSuite) TestListAndGetInstances() { + agentID := s.newAgent("list-instances") + other := s.newAgent("list-instances-other") + older, newer := uuid.New(), uuid.New() + report := applyReport(agentconfig.ModeApplySafe, baseConfig) + report.Warnings = []agentconfig.FieldError{{Path: "/x", Code: "unknown-field", Message: "m"}} + report.RemoteConfig = &agentconfig.RemoteConfig{Mode: agentconfig.ModeApplySafe} + s.Require().NoError(s.reportAt(s.svc, s.now.Add(-time.Hour), agentID, older, report)) + s.Require().NoError(s.reportAt(s.svc, s.now, agentID, newer, report)) + s.Require().NoError(s.svc.UpsertReport(s.ctx, other, nil, uuid.New(), report)) + + list, err := s.svc.ListInstances(s.ctx, agentID) + s.Require().NoError(err) + s.Require().Len(list, 2) + s.Equal(newer, list[0].InstanceID, "most recently seen first") + s.Equal(older, list[1].InstanceID) + for _, i := range list { + s.NotNil(i.ID) + s.Equal(agentID, i.AgentID) + s.Nil(i.BaseConfig, "heavy column not loaded") + s.Nil(i.EffectiveConfig, "heavy column not loaded") + s.NotNil(i.Warnings) + s.NotNil(i.RemoteConfig) + s.Equal("sha256:abc", *i.EffectiveDigest) + s.Equal(agentconfig.StatusApplied, i.ReportedStatus) + } + + full, err := s.svc.GetInstance(s.ctx, agentID, newer) + s.Require().NoError(err) + s.NotNil(full.BaseConfig) + s.NotNil(full.EffectiveConfig) + + _, err = s.svc.GetInstance(s.ctx, other, newer) + s.ErrorIs(err, agentcfg.ErrNotFound, "another agent's instance") + _, err = s.svc.GetInstance(s.ctx, agentID, uuid.New()) + s.ErrorIs(err, agentcfg.ErrNotFound) + + empty, err := s.svc.ListInstances(s.ctx, s.newAgent("no-instances")) + s.Require().NoError(err) + s.Empty(empty) +} + func (s *AgentCfgServiceIntegrationSuite) TestDeleteRevisionsForAgent() { agentA := s.newAgent("purge-a") agentB := s.newAgent("purge-b") diff --git a/internal/tests/migrate.go b/internal/tests/migrate.go index b6152090..f8b67b6c 100644 --- a/internal/tests/migrate.go +++ b/internal/tests/migrate.go @@ -174,6 +174,7 @@ func (t *TestMigrator) Up() error { &relational.AgentServiceAccountKey{}, &relational.AgentAuthEvent{}, &relational.AgentConfigRevision{}, + &relational.AgentInstance{}, &relational.SSOUserLink{}, &relational.SlackLinkAttempt{}, &relational.SlackUserLink{}, @@ -551,6 +552,7 @@ func (t *TestMigrator) Down() error { "poam_findings", "poam_risks", + &relational.AgentInstance{}, &relational.AgentConfigRevision{}, &relational.AgentAuthEvent{}, &relational.AgentServiceAccountKey{},