api: add opt-in instance retention pruner

CHANGELOG.md

@@ -9,6 +9,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 ### Security
 ### Added
 
+- **Optional instance retention:** Added `--instance-retention` (default off) to delete instances that have not checked in within a given duration, with dry-run, batched deletes, and `nebraska_instances_pruned_total`. See `docs/instance-retention.md`. ([#1603](https://github.com/flatcar/nebraska/pull/1603))
 - **Custom CA Certificate for TLS:** Added `--ca-file` flag to trust additional CA certificates for TLS verification (e.g., internal CA, Let's Encrypt staging). Applies to the OIDC provider client and the syncer. Supports multiple PEM-encoded certs, additive to system CAs. Also exposed as `config.caFile` in the Helm chart.
 - **OEM Attribute Capture:** Instances now store OEM and Aleph version information from Omaha update requests. ([#1286](https://github.com/flatcar/nebraska/pull/1286))
 - **Multi-Step Updates with Floor Packages:** Added support for mandatory intermediate update versions (floor packages) that clients must install before reaching the target version. This enables safe migration paths for breaking changes by ensuring clients update through specific versions in order. Floor packages can be configured per channel with optional reasons and are architecture-specific. ([#1195](https://github.com/flatcar/nebraska/pull/1195))

backend/pkg/api/runtime/prune.go

@@ -0,0 +1,107 @@
+package runtime
+
+import (
+	"context"
+	"fmt"
+	"time"
+)
+
+const (
+	defaultPruneBatchSize  = 500
+	defaultPruneMaxBatches = 20
+	defaultPruneTimeout    = 2 * time.Minute
+)
+
+// PruneStaleInstancesResult is the outcome of one prune pass.
+type PruneStaleInstancesResult struct {
+	// Candidates is the number of instances matching the cutoff. In a live
+	// run this equals Deleted; in a dry-run it is the would-delete count.
+	Candidates int64
+	// Deleted is the number of instance rows actually removed. Zero in dry-run.
+	Deleted int64
+	// MoreRemain is true when this pass hit the per-run batch cap and stale
+	// rows are likely still present for a later tick.
+	MoreRemain bool
+}
+
+// PruneStaleInstances deletes (or counts, when dryRun is true) instances whose
+// newest last_check_for_updates is older than cutoff. Instances that never
+// registered an application fall back to instance.created_ts.
+//
+// Related rows in instance_application, instance_status_history, event and
+// activity are removed by ON DELETE CASCADE. Each DELETE is limited to
+// batchSize rows, and a single call processes at most a handful of batches so
+// a first enable against years of dead nodes cannot lock the hot tables for
+// the whole hour.
+func (s *Service) PruneStaleInstances(cutoff time.Time, batchSize int, dryRun bool) (PruneStaleInstancesResult, error) {
+	return s.pruneStaleInstances(cutoff, batchSize, defaultPruneMaxBatches, dryRun)
+}
+
+func (s *Service) pruneStaleInstances(cutoff time.Time, batchSize, maxBatches int, dryRun bool) (PruneStaleInstancesResult, error) {
+	if batchSize <= 0 {
+		batchSize = defaultPruneBatchSize
+	}
+	if maxBatches <= 0 {
+		maxBatches = defaultPruneMaxBatches
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), defaultPruneTimeout)
+	defer cancel()
+
+	if dryRun {
+		var count int64
+		err := s.db.QueryRowxContext(ctx, `
+			SELECT COUNT(*) FROM (
+				SELECT i.id
+				FROM instance i
+				LEFT JOIN instance_application ia ON ia.instance_id = i.id
+				GROUP BY i.id, i.created_ts
+				HAVING COALESCE(MAX(ia.last_check_for_updates), i.created_ts) < $1
+			) stale
+		`, cutoff).Scan(&count)
+		if err != nil {
+			return PruneStaleInstancesResult{}, fmt.Errorf("count stale instances: %w", err)
+		}
+		return PruneStaleInstancesResult{Candidates: count}, nil
+	}
+
+	var deleted int64
+	var moreRemain bool
+	for batch := 0; batch < maxBatches; batch++ {
+		result, err := s.db.ExecContext(ctx, `
+			WITH stale AS (
+				SELECT i.id
+				FROM instance i
+				LEFT JOIN instance_application ia ON ia.instance_id = i.id
+				GROUP BY i.id, i.created_ts
+				HAVING COALESCE(MAX(ia.last_check_for_updates), i.created_ts) < $1
+				ORDER BY COALESCE(MAX(ia.last_check_for_updates), i.created_ts) ASC, i.id ASC
+				LIMIT $2
+			)
+			DELETE FROM instance i
+			USING stale
+			WHERE i.id = stale.id
+			  AND COALESCE(
+					(SELECT MAX(ia2.last_check_for_updates) FROM instance_application ia2 WHERE ia2.instance_id = i.id),
+					i.created_ts
+			  ) < $1
+		`, cutoff, batchSize)
+		if err != nil {
+			return PruneStaleInstancesResult{Candidates: deleted, Deleted: deleted}, fmt.Errorf("delete stale instances: %w", err)
+		}
+		n, err := result.RowsAffected()
+		if err != nil {
+			return PruneStaleInstancesResult{Candidates: deleted, Deleted: deleted}, fmt.Errorf("stale instance rows affected: %w", err)
+		}
+		deleted += n
+		if n < int64(batchSize) {
+			moreRemain = false
+			break
+		}
+		if batch == maxBatches-1 {
+			moreRemain = true
+		}
+	}
+
+	return PruneStaleInstancesResult{Candidates: deleted, Deleted: deleted, MoreRemain: moreRemain}, nil
+}

backend/pkg/api/runtime/prune_test.go

@@ -0,0 +1,188 @@
+package runtime
+
+import (
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/flatcar/nebraska/backend/pkg/api"
+	"github.com/flatcar/nebraska/backend/pkg/api/admin"
+	"github.com/flatcar/nebraska/backend/pkg/api/internal/dbconn"
+	"github.com/flatcar/nebraska/backend/pkg/api/types"
+)
+
+func setupPruneGroup(t *testing.T, a *api.API, name string) (*admin.Service, *Service, *types.Application, *types.Group) {
+	t.Helper()
+	as := adminSvc(a)
+	rs := runtimeSvc(a)
+	tTeam, err := as.AddTeam(&types.Team{Name: name + "_team"})
+	require.NoError(t, err)
+	tApp, err := as.AddApp(&types.Application{Name: name + "_app", TeamID: tTeam.ID})
+	require.NoError(t, err)
+	tGroup, err := as.AddGroup(&types.Group{Name: name + "_group", ApplicationID: tApp.ID, PolicyUpdatesEnabled: true, PolicyPeriodInterval: "15 minutes", PolicyMaxUpdatesPerPeriod: 2, PolicyUpdateTimeout: "60 minutes"})
+	require.NoError(t, err)
+	return as, rs, tApp, tGroup
+}
+
+func backdateCheckIn(t *testing.T, a *api.API, instanceID string, age time.Duration) {
+	t.Helper()
+	_, err := dbconn.DB(a.Conn()).Exec(
+		`UPDATE instance_application SET last_check_for_updates = $1 WHERE instance_id = $2`,
+		time.Now().UTC().Add(-age),
+		instanceID,
+	)
+	require.NoError(t, err)
+}
+
+func TestPruneStaleInstancesDeletesOldAndKeepsFresh(t *testing.T) {
+	a := newForTest(t)
+	defer a.Close()
+	_, rs, tApp, tGroup := setupPruneGroup(t, a, "prune")
+
+	stale, err := rs.RegisterInstance(types.Instance{ID: uuid.New().String(), IP: "10.0.0.1"}, NewInstanceApplication(tApp.ID, tGroup.ID, "1.0.0"))
+	require.NoError(t, err)
+	fresh, err := rs.RegisterInstance(types.Instance{ID: uuid.New().String(), IP: "10.0.0.2"}, NewInstanceApplication(tApp.ID, tGroup.ID, "1.0.0"))
+	require.NoError(t, err)
+
+	require.NoError(t, rs.grantUpdate(stale, "1.0.1"))
+	require.NoError(t, rs.updateInstanceStatus(stale.ID, tApp.ID, types.InstanceStatusComplete))
+	backdateCheckIn(t, a, stale.ID, 2*time.Hour)
+
+	result, err := rs.PruneStaleInstances(time.Now().UTC().Add(-time.Hour), 500, false)
+	require.NoError(t, err)
+	assert.Equal(t, int64(1), result.Deleted)
+	assert.False(t, result.MoreRemain)
+
+	_, err = a.GetInstance(stale.ID, tApp.ID)
+	assert.Error(t, err, "stale instance should be gone")
+
+	stillThere, err := a.GetInstance(fresh.ID, tApp.ID)
+	require.NoError(t, err)
+	assert.Equal(t, fresh.ID, stillThere.ID)
+
+	history, err := a.GetInstanceStatusHistory(stale.ID, tApp.ID, tGroup.ID, 100)
+	require.NoError(t, err)
+	assert.Empty(t, history, "status history should cascade-delete with the instance")
+}
+
+func TestPruneStaleInstancesDryRunDoesNotDelete(t *testing.T) {
+	a := newForTest(t)
+	defer a.Close()
+	_, rs, tApp, tGroup := setupPruneGroup(t, a, "prune_dry")
+
+	stale, err := rs.RegisterInstance(types.Instance{ID: uuid.New().String(), IP: "10.0.0.3"}, NewInstanceApplication(tApp.ID, tGroup.ID, "1.0.0"))
+	require.NoError(t, err)
+	backdateCheckIn(t, a, stale.ID, 2*time.Hour)
+
+	result, err := rs.PruneStaleInstances(time.Now().UTC().Add(-time.Hour), 500, true)
+	require.NoError(t, err)
+	assert.Equal(t, int64(1), result.Candidates)
+	assert.Equal(t, int64(0), result.Deleted)
+
+	_, err = a.GetInstance(stale.ID, tApp.ID)
+	require.NoError(t, err, "dry-run must not delete the instance")
+}
+
+func TestPruneStaleInstancesKeepsInstanceWithAnyFreshCheckIn(t *testing.T) {
+	a := newForTest(t)
+	defer a.Close()
+	as, rs, tApp1, tGroup1 := setupPruneGroup(t, a, "prune_multi")
+	tApp2, err := as.AddApp(&types.Application{Name: "prune_multi_app2", TeamID: tApp1.TeamID})
+	require.NoError(t, err)
+	tGroup2, err := as.AddGroup(&types.Group{Name: "prune_multi_g2", ApplicationID: tApp2.ID, PolicyUpdatesEnabled: true, PolicyPeriodInterval: "15 minutes", PolicyMaxUpdatesPerPeriod: 2, PolicyUpdateTimeout: "60 minutes"})
+	require.NoError(t, err)
+
+	id := uuid.New().String()
+	_, err = rs.RegisterInstance(types.Instance{ID: id, IP: "10.0.0.4"}, NewInstanceApplication(tApp1.ID, tGroup1.ID, "1.0.0"))
+	require.NoError(t, err)
+	_, err = rs.RegisterInstance(types.Instance{ID: id, IP: "10.0.0.4"}, NewInstanceApplication(tApp2.ID, tGroup2.ID, "1.0.0"))
+	require.NoError(t, err)
+
+	_, err = dbconn.DB(a.Conn()).Exec(
+		`UPDATE instance_application SET last_check_for_updates = $1 WHERE instance_id = $2 AND application_id = $3`,
+		time.Now().UTC().Add(-24*time.Hour),
+		id,
+		tApp1.ID,
+	)
+	require.NoError(t, err)
+
+	result, err := rs.PruneStaleInstances(time.Now().UTC().Add(-time.Hour), 500, false)
+	require.NoError(t, err)
+
+	_, err = a.GetInstance(id, tApp2.ID)
+	require.NoError(t, err, "instance must stay if any application still checks in")
+	assert.Equal(t, int64(0), result.Deleted)
+}
+
+func TestPruneStaleInstancesBatches(t *testing.T) {
+	a := newForTest(t)
+	defer a.Close()
+	_, rs, tApp, tGroup := setupPruneGroup(t, a, "prune_batch")
+
+	ids := make([]string, 3)
+	for i := range ids {
+		inst, err := rs.RegisterInstance(types.Instance{ID: uuid.New().String(), IP: "10.0.0.10"}, NewInstanceApplication(tApp.ID, tGroup.ID, "1.0.0"))
+		require.NoError(t, err)
+		ids[i] = inst.ID
+		backdateCheckIn(t, a, inst.ID, 2*time.Hour)
+	}
+
+	result, err := rs.PruneStaleInstances(time.Now().UTC().Add(-time.Hour), 2, false)
+	require.NoError(t, err)
+	assert.Equal(t, int64(3), result.Deleted)
+
+	for _, id := range ids {
+		_, err := a.GetInstance(id, tApp.ID)
+		assert.Error(t, err, "batched prune should remove all stale instances")
+	}
+}
+
+func TestPruneStaleInstancesRespectsMaxBatches(t *testing.T) {
+	a := newForTest(t)
+	defer a.Close()
+	_, rs, tApp, tGroup := setupPruneGroup(t, a, "prune_cap")
+
+	ids := make([]string, 3)
+	for i := range ids {
+		inst, err := rs.RegisterInstance(types.Instance{ID: uuid.New().String(), IP: "10.0.0.11"}, NewInstanceApplication(tApp.ID, tGroup.ID, "1.0.0"))
+		require.NoError(t, err)
+		ids[i] = inst.ID
+		backdateCheckIn(t, a, inst.ID, 2*time.Hour)
+	}
+
+	result, err := rs.pruneStaleInstances(time.Now().UTC().Add(-time.Hour), 1, 2, false)
+	require.NoError(t, err)
+	assert.Equal(t, int64(2), result.Deleted)
+	assert.True(t, result.MoreRemain)
+
+	remaining := 0
+	for _, id := range ids {
+		if _, err := a.GetInstance(id, tApp.ID); err == nil {
+			remaining++
+		}
+	}
+	assert.Equal(t, 1, remaining)
+}
+
+func TestPruneStaleInstancesOrphanCreatedTs(t *testing.T) {
+	a := newForTest(t)
+	defer a.Close()
+	rs := runtimeSvc(a)
+	db := dbconn.DB(a.Conn())
+
+	orphanID := uuid.New().String()
+	_, err := db.Exec(`INSERT INTO instance (id, ip, created_ts) VALUES ($1, $2, $3)`, orphanID, "10.0.0.9", time.Now().UTC().Add(-3*time.Hour))
+	require.NoError(t, err)
+
+	result, err := rs.PruneStaleInstances(time.Now().UTC().Add(-time.Hour), 500, false)
+	require.NoError(t, err)
+	assert.Equal(t, int64(1), result.Deleted)
+
+	var n int
+	err = db.QueryRowx(`SELECT COUNT(*) FROM instance WHERE id = $1`, orphanID).Scan(&n)
+	require.NoError(t, err)
+	assert.Equal(t, 0, n)
+}

backend/pkg/config/config.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"net/url"
 	"os"
+	"time"
 
 	"github.com/knadh/koanf"
 	"github.com/knadh/koanf/providers/basicflag"
@@ -54,6 +55,14 @@ type Config struct {
 	OidcUseUserInfo   bool   `koanf:"oidc-use-userinfo"`
 	CAFile            string `koanf:"ca-file"`
 	CACertPool        *x509.CertPool
+
+	// InstanceRetention, when greater than zero, enables a background job that
+	// deletes instances whose newest last_check_for_updates (or created_ts if
+	// they never checked in) is older than this duration. Zero disables it so
+	// existing deployments do not change behaviour on upgrade.
+	InstanceRetention          time.Duration `koanf:"instance-retention"`
+	InstanceRetentionDryRun    bool          `koanf:"instance-retention-dry-run"`
+	InstanceRetentionBatchSize uint          `koanf:"instance-retention-batch-size"`
 }
 
 const (
@@ -100,6 +109,16 @@ func (c *Config) Validate() error {
 		}
 	}
 
+	if c.InstanceRetention < 0 {
+		return errors.New("instance-retention must be zero or a positive duration")
+	}
+	if c.InstanceRetention > 0 && c.InstanceRetentionBatchSize == 0 {
+		return errors.New("instance-retention-batch-size must be greater than zero when instance-retention is enabled")
+	}
+	if c.InstanceRetentionBatchSize > 10000 {
+		return errors.New("instance-retention-batch-size must be at most 10000")
+	}
+
 	return nil
 }
 
@@ -145,6 +164,9 @@ func Parse() (*Config, error) {
 	f.String("api-endpoint-suffix", "", "Additional suffix for the API endpoint to serve Omaha clients on; use a secret to only serve your clients, e.g., mysecret results in /v1/update/mysecret")
 	f.Bool("debug", false, "sets log level to debug")
 	f.Uint("port", 8000, "port to run server")
+	f.Duration("instance-retention", 0, "delete instances whose newest last_check_for_updates (or created_ts if they never checked in) is older than this duration; 0 disables the pruner")
+	f.Bool("instance-retention-dry-run", false, "log how many instances would be deleted by instance-retention without deleting them")
+	f.Uint("instance-retention-batch-size", 500, "maximum number of instances deleted per batch when instance-retention is enabled")
 
 	k := koanf.New(".")
 

backend/pkg/config/config_test.go

@@ -0,0 +1,35 @@
+package config
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestValidateInstanceRetention(t *testing.T) {
+	t.Run("disabled by default", func(t *testing.T) {
+		c := &Config{AuthMode: "noop"}
+		assert.NoError(t, c.Validate())
+	})
+
+	t.Run("enabled requires batch size", func(t *testing.T) {
+		c := &Config{AuthMode: "noop", InstanceRetention: time.Hour}
+		assert.Error(t, c.Validate())
+	})
+
+	t.Run("enabled with batch size", func(t *testing.T) {
+		c := &Config{AuthMode: "noop", InstanceRetention: time.Hour, InstanceRetentionBatchSize: 500}
+		assert.NoError(t, c.Validate())
+	})
+
+	t.Run("negative retention", func(t *testing.T) {
+		c := &Config{AuthMode: "noop", InstanceRetention: -time.Hour, InstanceRetentionBatchSize: 500}
+		assert.Error(t, c.Validate())
+	})
+
+	t.Run("batch size too large", func(t *testing.T) {
+		c := &Config{AuthMode: "noop", InstanceRetention: time.Hour, InstanceRetentionBatchSize: 10001}
+		assert.Error(t, c.Validate())
+	})
+}