diff --git a/internal/ingest/run.go b/internal/ingest/run.go index 58a1579..9046898 100644 --- a/internal/ingest/run.go +++ b/internal/ingest/run.go @@ -17,6 +17,8 @@ type Deps struct { Store store.Backend } +var lastMaintenance time.Time + // Run blocks until ctx is cancelled. func Run(ctx context.Context, deps *Deps) { cfg := config.Load() @@ -34,6 +36,10 @@ func Run(ctx context.Context, deps *Deps) { log.Printf("evobgp-ingest: stopped") return case <-t.C: + if deps.Store != nil && time.Since(lastMaintenance) > time.Hour { + deps.Store.RunPeriodicMaintenance(ctx) + lastMaintenance = time.Now() + } prefetchCtx, cancel := context.WithTimeout(ctx, 50*time.Second) err := pipeline.PrefetchCDNSourceETags(prefetchCtx, deps.Store, hc) cancel() diff --git a/internal/repository/asn_cache.go b/internal/repository/asn_cache.go index f6d106f..1b43aae 100644 --- a/internal/repository/asn_cache.go +++ b/internal/repository/asn_cache.go @@ -11,14 +11,22 @@ import ( "github.com/jackc/pgx/v5" ) +func asnCacheRowTableExists(ctx context.Context, q queryRower) bool { + var n int + err := q.QueryRow(ctx, ` + SELECT 1 FROM information_schema.tables + WHERE table_schema = 'public' AND table_name = 'asn_prefix_cache_row' + LIMIT 1`).Scan(&n) + return err == nil +} + func (p *Postgres) GetASNPrefixCache(asn int64) (*store.ASNPrefixCacheEntry, bool, error) { ctx := context.Background() var holder string var fetchedAt time.Time - var raw []byte err := p.pool.QueryRow(ctx, ` - SELECT holder, fetched_at, prefixes_json FROM asn_prefix_cache WHERE asn = $1`, asn). - Scan(&holder, &fetchedAt, &raw) + SELECT holder, fetched_at FROM asn_prefix_cache WHERE asn = $1`, asn). + Scan(&holder, &fetchedAt) if err != nil { if errors.Is(err, pgx.ErrNoRows) { return nil, false, nil @@ -26,8 +34,25 @@ func (p *Postgres) GetASNPrefixCache(asn int64) (*store.ASNPrefixCacheEntry, boo return nil, false, err } var prefixes []string - if len(raw) > 0 { - _ = json.Unmarshal(raw, &prefixes) + if asnCacheRowTableExists(ctx, p.pool) { + rows, qerr := p.pool.Query(ctx, ` + SELECT prefix::text FROM asn_prefix_cache_row WHERE asn = $1 ORDER BY prefix`, asn) + if qerr != nil { + return nil, false, qerr + } + defer rows.Close() + for rows.Next() { + var s string + if err := rows.Scan(&s); err != nil { + continue + } + prefixes = append(prefixes, s) + } + } else { + var raw []byte + if err := p.pool.QueryRow(ctx, `SELECT prefixes_json FROM asn_prefix_cache WHERE asn = $1`, asn).Scan(&raw); err == nil && len(raw) > 0 { + _ = json.Unmarshal(raw, &prefixes) + } } return &store.ASNPrefixCacheEntry{ ASN: asn, @@ -38,18 +63,40 @@ func (p *Postgres) GetASNPrefixCache(asn int64) (*store.ASNPrefixCacheEntry, boo } func (p *Postgres) SetASNPrefixCache(asn int64, holder string, prefixes []string) error { - raw, err := json.Marshal(prefixes) + ctx := context.Background() + tx, err := p.pool.Begin(ctx) if err != nil { return err } - ctx := context.Background() - _, err = p.pool.Exec(ctx, ` - INSERT INTO asn_prefix_cache (asn, holder, prefixes_json, fetched_at) - VALUES ($1, $2, $3::jsonb, now()) + defer func() { _ = tx.Rollback(ctx) }() + _, err = tx.Exec(ctx, ` + INSERT INTO asn_prefix_cache (asn, holder, fetched_at) + VALUES ($1, $2, now()) ON CONFLICT (asn) DO UPDATE SET holder = EXCLUDED.holder, - prefixes_json = EXCLUDED.prefixes_json, - fetched_at = EXCLUDED.fetched_at`, - asn, holder, string(raw)) - return err + fetched_at = EXCLUDED.fetched_at`, asn, holder) + if err != nil { + return err + } + if asnCacheRowTableExists(ctx, tx) { + if _, err := tx.Exec(ctx, `DELETE FROM asn_prefix_cache_row WHERE asn = $1`, asn); err != nil { + return err + } + for _, pfx := range prefixes { + if _, err := tx.Exec(ctx, ` + INSERT INTO asn_prefix_cache_row (asn, prefix) VALUES ($1, $2::cidr)`, asn, pfx); err != nil { + return err + } + } + } else { + raw, err := json.Marshal(prefixes) + if err != nil { + return err + } + if _, err := tx.Exec(ctx, ` + UPDATE asn_prefix_cache SET prefixes_json = $2::jsonb WHERE asn = $1`, asn, string(raw)); err != nil { + return err + } + } + return tx.Commit(ctx) } diff --git a/internal/repository/maintenance.go b/internal/repository/maintenance.go new file mode 100644 index 0000000..808f32d --- /dev/null +++ b/internal/repository/maintenance.go @@ -0,0 +1,29 @@ +package repository + +import ( + "context" + "time" +) + +const ( + jobAuditRetentionDays = 90 + asnCacheRetentionDays = 7 +) + +// RunPeriodicMaintenance prunes stale job_audit and asn_prefix_cache rows (PostgreSQL). +func (p *Postgres) RunPeriodicMaintenance(ctx context.Context) { + if p == nil || p.pool == nil { + return + } + if ctx == nil { + ctx = context.Background() + } + jobCutoff := time.Now().UTC().Add(-time.Duration(jobAuditRetentionDays) * 24 * time.Hour) + _, _ = p.pool.Exec(ctx, ` + DELETE FROM job_audit + WHERE created_at < $1 + AND status IN ('succeeded', 'failed', 'cancelled')`, jobCutoff) + asnCutoff := time.Now().UTC().Add(-time.Duration(asnCacheRetentionDays) * 24 * time.Hour) + _, _ = p.pool.Exec(ctx, ` + DELETE FROM asn_prefix_cache WHERE fetched_at < $1`, asnCutoff) +} diff --git a/internal/store/backend.go b/internal/store/backend.go index 35b9a01..7208093 100644 --- a/internal/store/backend.go +++ b/internal/store/backend.go @@ -112,6 +112,9 @@ type Backend interface { // Ping verifies backend connectivity (no-op for in-memory). Ping(ctx context.Context) error + + // RunPeriodicMaintenance prunes stale DB rows (no-op for in-memory). + RunPeriodicMaintenance(ctx context.Context) } // ASNPrefixCacheEntry is a cached RIPEstat response for one ASN. diff --git a/internal/store/memory.go b/internal/store/memory.go index 8b31954..8aa1955 100644 --- a/internal/store/memory.go +++ b/internal/store/memory.go @@ -305,6 +305,11 @@ func (m *Memory) Ping(ctx context.Context) error { return nil } +// RunPeriodicMaintenance is a no-op for the in-memory backend. +func (m *Memory) RunPeriodicMaintenance(ctx context.Context) { + _ = ctx +} + // ListTenantIDs returns tenant ids sorted lexicographically. func (m *Memory) ListTenantIDs() ([]string, error) { m.mu.RLock() diff --git a/migrations/postgres/000020_asn_cache_rows.down.sql b/migrations/postgres/000020_asn_cache_rows.down.sql new file mode 100644 index 0000000..d3e9d9c --- /dev/null +++ b/migrations/postgres/000020_asn_cache_rows.down.sql @@ -0,0 +1,10 @@ +ALTER TABLE asn_prefix_cache ADD COLUMN prefixes_json JSONB NOT NULL DEFAULT '[]'; + +UPDATE asn_prefix_cache apc +SET prefixes_json = COALESCE(( + SELECT jsonb_agg(apcr.prefix::text ORDER BY apcr.prefix::text) + FROM asn_prefix_cache_row apcr + WHERE apcr.asn = apc.asn +), '[]'::jsonb); + +DROP TABLE IF EXISTS asn_prefix_cache_row; diff --git a/migrations/postgres/000020_asn_cache_rows.up.sql b/migrations/postgres/000020_asn_cache_rows.up.sql new file mode 100644 index 0000000..759231d --- /dev/null +++ b/migrations/postgres/000020_asn_cache_rows.up.sql @@ -0,0 +1,14 @@ +CREATE TABLE asn_prefix_cache_row ( + asn BIGINT NOT NULL REFERENCES asn_prefix_cache (asn) ON DELETE CASCADE, + prefix CIDR NOT NULL, + PRIMARY KEY (asn, prefix) +); + +INSERT INTO asn_prefix_cache_row (asn, prefix) +SELECT apc.asn, t.elem::cidr +FROM asn_prefix_cache apc +CROSS JOIN LATERAL jsonb_array_elements_text(apc.prefixes_json) AS t(elem) +WHERE jsonb_typeof(apc.prefixes_json) = 'array' + AND jsonb_array_length(apc.prefixes_json) > 0; + +ALTER TABLE asn_prefix_cache DROP COLUMN prefixes_json; diff --git a/migrations/sqlite/000020_asn_cache_rows.down.sql b/migrations/sqlite/000020_asn_cache_rows.down.sql new file mode 100644 index 0000000..553bfe5 --- /dev/null +++ b/migrations/sqlite/000020_asn_cache_rows.down.sql @@ -0,0 +1,2 @@ +ALTER TABLE asn_prefix_cache ADD COLUMN prefixes_json TEXT NOT NULL DEFAULT '[]'; +DROP TABLE IF EXISTS asn_prefix_cache_row; diff --git a/migrations/sqlite/000020_asn_cache_rows.up.sql b/migrations/sqlite/000020_asn_cache_rows.up.sql new file mode 100644 index 0000000..aa8fdf9 --- /dev/null +++ b/migrations/sqlite/000020_asn_cache_rows.up.sql @@ -0,0 +1,7 @@ +CREATE TABLE asn_prefix_cache_row ( + asn INTEGER NOT NULL REFERENCES asn_prefix_cache (asn) ON DELETE CASCADE, + prefix TEXT NOT NULL, + PRIMARY KEY (asn, prefix) +); + +ALTER TABLE asn_prefix_cache DROP COLUMN prefixes_json;