Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
331 changes: 0 additions & 331 deletions internal/timeline/sqlite_store_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -1148,215 +1148,6 @@ func TestSQLiteStore_Migration_AddsClusterContext(t *testing.T) {
}
}

// A relist re-emits the same informer id; INSERT ... ON CONFLICT must leave the
// original row untouched (no k8s_event mutation path) instead of duplicating it.
// Mirrors TestMemoryStore_DedupesIdenticalInformerID.
func TestSQLiteStore_DedupesIdenticalInformerID(t *testing.T) {
store, cleanup := createTestSQLiteStore(t)
defer cleanup()
ctx := context.Background()

rv := "100"
add := NewInformerEvent("Deployment", "apps/v1", "team-a", "web", "uid-1", rv, EventTypeAdd, HealthHealthy, nil, nil, nil, nil)
relistUpdate := NewInformerEvent("Deployment", "apps/v1", "team-a", "web", "uid-1", rv, EventTypeUpdate, HealthHealthy, nil, nil, nil, nil)
if add.ID != relistUpdate.ID {
t.Fatalf("relist add/update must share an id: %q vs %q", add.ID, relistUpdate.ID)
}

if err := store.Append(ctx, add); err != nil {
t.Fatalf("Append add: %v", err)
}
if err := store.Append(ctx, relistUpdate); err != nil {
t.Fatalf("Append relistUpdate: %v", err)
}

got, err := store.Query(ctx, QueryOptions{Limit: 10, IncludeManaged: true})
if err != nil {
t.Fatalf("Query: %v", err)
}
if len(got) != 1 {
t.Fatalf("expected 1 deduped row, got %d: %+v", len(got), got)
}
// The original add is kept — the relist did NOT overwrite it into an update.
if got[0].EventType != EventTypeAdd {
t.Fatalf("informer relist must not mutate the row, got event_type %q want %q", got[0].EventType, EventTypeAdd)
}

del := NewInformerEvent("Deployment", "apps/v1", "team-a", "web", "uid-1", rv, EventTypeDelete, HealthUnknown, nil, nil, nil, nil)
if del.ID == add.ID {
t.Fatalf("delete must get a distinct id, got %q for both", del.ID)
}
if err := store.Append(ctx, del); err != nil {
t.Fatalf("Append del: %v", err)
}
got, err = store.Query(ctx, QueryOptions{Limit: 10, IncludeManaged: true})
if err != nil {
t.Fatalf("Query: %v", err)
}
if len(got) != 2 {
t.Fatalf("expected add + delete = 2 rows, got %d", len(got))
}
}

// A K8s Event's count bump reuses the uid-based id; the store must upsert the
// mutable fields (count/message/timestamp) in place, not drop the bump or append
// a duplicate. Mirrors TestMemoryStore_K8sEventCountBumpUpserts.
func TestSQLiteStore_K8sEventCountBumpUpserts(t *testing.T) {
store, cleanup := createTestSQLiteStore(t)
defer cleanup()
ctx := context.Background()

base := time.Now()
mk := func(count int32, message string, ts time.Time) TimelineEvent {
return TimelineEvent{
ID: "k8s-uid-1", Timestamp: ts, Source: SourceK8sEvent,
Kind: "Pod", Namespace: "team-a", Name: "web-abc",
EventType: EventTypeWarning, Reason: "BackOff", Message: message, Count: count,
}
}
if err := store.Append(ctx, mk(1, "back-off 10s", base)); err != nil {
t.Fatalf("Append first: %v", err)
}
if err := store.Append(ctx, mk(5, "back-off 40s", base.Add(30*time.Second))); err != nil {
t.Fatalf("Append bump: %v", err)
}

got, err := store.Query(ctx, QueryOptions{Limit: 10, IncludeManaged: true, IncludeK8sEvents: true})
if err != nil {
t.Fatalf("Query: %v", err)
}
if len(got) != 1 {
t.Fatalf("expected 1 upserted row, got %d: %+v", len(got), got)
}
if got[0].Count != 5 {
t.Fatalf("expected refreshed count 5, got %d", got[0].Count)
}
if got[0].Message != "back-off 40s" {
t.Fatalf("expected refreshed message, got %q", got[0].Message)
}
if !got[0].Timestamp.Equal(base.Add(30 * time.Second)) {
t.Fatalf("expected refreshed timestamp %v, got %v", base.Add(30*time.Second), got[0].Timestamp)
}
}

// An out-of-order older revision of the same K8s Event uid must NOT clobber the
// newer row already stored. Mirrors the recency guard in
// TestMemoryStore_K8sEventBumpMovesToRecency.
func TestSQLiteStore_K8sEventStaleBumpIgnored(t *testing.T) {
store, cleanup := createTestSQLiteStore(t)
defer cleanup()
ctx := context.Background()

base := time.Now()
mk := func(count int32, ts time.Time) TimelineEvent {
return TimelineEvent{
ID: "k8s-uid-1", Timestamp: ts, Source: SourceK8sEvent,
Kind: "Pod", Namespace: "team-a", Name: "web-abc",
EventType: EventTypeWarning, Reason: "BackOff", Count: count,
}
}
// Newest revision arrives first, then a stale older relay of the same uid.
if err := store.Append(ctx, mk(5, base.Add(30*time.Second))); err != nil {
t.Fatalf("Append newer: %v", err)
}
if err := store.Append(ctx, mk(1, base)); err != nil {
t.Fatalf("Append stale: %v", err)
}

got, err := store.Query(ctx, QueryOptions{Limit: 10, IncludeManaged: true, IncludeK8sEvents: true})
if err != nil {
t.Fatalf("Query: %v", err)
}
if len(got) != 1 {
t.Fatalf("expected 1 row, got %d: %+v", len(got), got)
}
if got[0].Count != 5 {
t.Fatalf("stale relay must not clobber newer row, got count %d want 5", got[0].Count)
}
if !got[0].Timestamp.Equal(base.Add(30 * time.Second)) {
t.Fatalf("stale relay must not roll back timestamp, got %v", got[0].Timestamp)
}
}

// Seq is the delta cursor: SinceSeq must return exactly what arrived after the
// cursor, keyed on arrival order — a late event with an older timestamp still
// lands ahead of the cursor. Mirrors TestMemoryStore_SeqCursor.
func TestSQLiteStore_SeqCursor(t *testing.T) {
store, cleanup := createTestSQLiteStore(t)
defer cleanup()
ctx := context.Background()

base := time.Now()
mk := func(id string, ts time.Time) TimelineEvent {
return TimelineEvent{
ID: id, Timestamp: ts, Source: SourceInformer,
Kind: "Pod", Namespace: "default", Name: id, EventType: EventTypeUpdate,
}
}
if err := store.AppendBatch(ctx, []TimelineEvent{mk("a", base), mk("b", base.Add(time.Second))}); err != nil {
t.Fatalf("AppendBatch: %v", err)
}
all, err := store.Query(ctx, QueryOptions{Limit: 10, IncludeManaged: true})
if err != nil {
t.Fatalf("Query: %v", err)
}
if len(all) != 2 || all[0].Seq == 0 || all[1].Seq == 0 || all[0].Seq == all[1].Seq {
t.Fatalf("expected 2 events with distinct non-zero seqs, got %+v", all)
}
cursor := max(all[0].Seq, all[1].Seq)

if err := store.Append(ctx, mk("late", base.Add(-time.Hour))); err != nil {
t.Fatalf("Append late: %v", err)
}
delta, err := store.Query(ctx, QueryOptions{Limit: 10, IncludeManaged: true, SinceSeq: cursor})
if err != nil {
t.Fatalf("Query delta: %v", err)
}
if len(delta) != 1 || delta[0].ID != "late" {
t.Fatalf("expected exactly the late arrival past the cursor, got %+v", delta)
}
if delta[0].Seq <= cursor {
t.Fatalf("late arrival seq %d must exceed cursor %d", delta[0].Seq, cursor)
}
}

// A K8s Event count bump upserts in place but must take a fresh seq, or a
// delta reader holding a cursor past the original arrival never sees the bump.
// Mirrors TestMemoryStore_K8sEventBumpAdvancesSeq.
func TestSQLiteStore_K8sEventBumpAdvancesSeq(t *testing.T) {
store, cleanup := createTestSQLiteStore(t)
defer cleanup()
ctx := context.Background()

base := time.Now()
mk := func(count int32, ts time.Time) TimelineEvent {
return TimelineEvent{
ID: "k8s-uid-1", Timestamp: ts, Source: SourceK8sEvent,
Kind: "Pod", Namespace: "default", Name: "web-abc",
EventType: EventTypeWarning, Reason: "BackOff", Count: count,
}
}
if err := store.Append(ctx, mk(1, base)); err != nil {
t.Fatalf("Append first: %v", err)
}
first, err := store.Query(ctx, QueryOptions{Limit: 10, IncludeManaged: true, IncludeK8sEvents: true})
if err != nil || len(first) != 1 {
t.Fatalf("Query first: %v %+v", err, first)
}
cursor := first[0].Seq

if err := store.Append(ctx, mk(5, base.Add(30*time.Second))); err != nil {
t.Fatalf("Append bump: %v", err)
}
delta, err := store.Query(ctx, QueryOptions{Limit: 10, IncludeManaged: true, IncludeK8sEvents: true, SinceSeq: cursor})
if err != nil {
t.Fatalf("Query delta: %v", err)
}
if len(delta) != 1 || delta[0].Count != 5 {
t.Fatalf("expected the bumped row past the cursor, got %+v", delta)
}
}

// A database created before the seq column existed must come back with arrival
// numbers backfilled from rowid, so pre-upgrade rows are addressable by cursor.
func TestSQLiteStore_SeqMigrationBackfillsFromRowid(t *testing.T) {
Expand Down Expand Up @@ -1452,76 +1243,6 @@ func TestSQLiteStore_ResourceCreatedAtRoundtrip(t *testing.T) {
}
}

// A bump that lost its enrichment (tombstone expired) must not erase the
// owner/labels/createdAt the row already knows; a bump that carries fresh
// enrichment wins. Mirrors TestMemoryStore_K8sEventBumpPreservesEnrichment.
func TestSQLiteStore_K8sEventBumpPreservesEnrichment(t *testing.T) {
store, cleanup := createTestSQLiteStore(t)
defer cleanup()
ctx := context.Background()

base := time.Now()
born := base.Add(-time.Hour).Truncate(time.Millisecond)
if err := store.Append(ctx, TimelineEvent{
ID: "k8s-uid-1", Timestamp: base, Source: SourceK8sEvent,
Kind: "Pod", Namespace: "default", Name: "web-abc",
EventType: EventTypeWarning, Reason: "BackOff", Count: 1,
CreatedAt: &born,
Owner: &OwnerInfo{Kind: "ReplicaSet", Name: "web"},
Labels: map[string]string{"app": "web"},
}); err != nil {
t.Fatalf("Append enriched: %v", err)
}
if err := store.Append(ctx, TimelineEvent{
ID: "k8s-uid-1", Timestamp: base.Add(30 * time.Second), Source: SourceK8sEvent,
Kind: "Pod", Namespace: "default", Name: "web-abc",
EventType: EventTypeWarning, Reason: "BackOff", Count: 5,
}); err != nil {
t.Fatalf("Append bare bump: %v", err)
}

got, err := store.Query(ctx, QueryOptions{Limit: 10, IncludeManaged: true, IncludeK8sEvents: true})
if err != nil || len(got) != 1 {
t.Fatalf("Query: %v %+v", err, got)
}
if got[0].Count != 5 {
t.Fatalf("expected bumped count 5, got %d", got[0].Count)
}
if got[0].CreatedAt == nil || !got[0].CreatedAt.Equal(born) {
t.Fatalf("bare bump erased CreatedAt: %+v", got[0].CreatedAt)
}
if got[0].Owner == nil || got[0].Owner.Name != "web" {
t.Fatalf("bare bump erased Owner: %+v", got[0].Owner)
}
if got[0].Labels["app"] != "web" {
t.Fatalf("bare bump erased Labels: %+v", got[0].Labels)
}

// The inverse: a bump that carries enrichment fills a row that lacked it.
if err := store.Append(ctx, TimelineEvent{
ID: "k8s-uid-2", Timestamp: base, Source: SourceK8sEvent,
Kind: "Pod", Namespace: "default", Name: "late-enriched",
EventType: EventTypeWarning, Reason: "BackOff", Count: 1,
}); err != nil {
t.Fatalf("Append bare first: %v", err)
}
if err := store.Append(ctx, TimelineEvent{
ID: "k8s-uid-2", Timestamp: base.Add(time.Minute), Source: SourceK8sEvent,
Kind: "Pod", Namespace: "default", Name: "late-enriched",
EventType: EventTypeWarning, Reason: "BackOff", Count: 2,
Owner: &OwnerInfo{Kind: "Job", Name: "batch"},
}); err != nil {
t.Fatalf("Append enriched bump: %v", err)
}
row, err := store.GetEvent(ctx, "k8s-uid-2")
if err != nil || row == nil {
t.Fatalf("GetEvent: %v %+v", err, row)
}
if row.Owner == nil || row.Owner.Name != "batch" {
t.Fatalf("enriched bump did not fill Owner: %+v", row.Owner)
}
}

// Timestamps are stored as TEXT and ordered lexically; informer events carry
// the host's local zone while K8s events are UTC. Without UTC normalization on
// write, a local-zone wall clock sorts ahead of an earlier-instant UTC string.
Expand Down Expand Up @@ -1559,58 +1280,6 @@ func TestSQLiteStore_UTCTimestampOrdering(t *testing.T) {
}
}

// Delta reads must page by ascending arrival order so a burst larger than the
// page limit isn't skipped as the server advances the cursor by each page's max
// seq. Successive polls must cover every unseen event with no gaps.
func TestSQLiteStore_DeltaPagesAscendingUnderLimit(t *testing.T) {
store, cleanup := createTestSQLiteStore(t)
defer cleanup()
ctx := context.Background()

base := time.Now()
for i := 0; i < 6; i++ {
ev := TimelineEvent{
ID: fmt.Sprintf("e%d", i), Timestamp: base.Add(time.Duration(i) * time.Second),
Source: SourceInformer, Kind: "Deployment", Namespace: "default",
Name: fmt.Sprintf("e%d", i), EventType: EventTypeUpdate,
}
if err := store.Append(ctx, ev); err != nil {
t.Fatalf("Append %d: %v", i, err)
}
}

cursor := int64(1)
seen := map[int64]bool{}
for polls := 0; polls < 10; polls++ {
page, err := store.Query(ctx, QueryOptions{Limit: 2, IncludeManaged: true, SinceSeq: cursor})
if err != nil {
t.Fatalf("delta query: %v", err)
}
if len(page) == 0 {
break
}
if len(page) == 2 && page[0].Seq >= page[1].Seq {
t.Fatalf("delta page not ascending by seq: %d,%d", page[0].Seq, page[1].Seq)
}
var maxSeq int64
for _, e := range page {
if seen[e.Seq] {
t.Fatalf("seq %d returned twice across polls", e.Seq)
}
seen[e.Seq] = true
if e.Seq > maxSeq {
maxSeq = e.Seq
}
}
cursor = maxSeq
}
for want := int64(2); want <= 6; want++ {
if !seen[want] {
t.Fatalf("delta paging skipped seq %d; seen=%v", want, seen)
}
}
}

// seq must not rewind when retention empties the table: a live client's cursor
// (seq>N) would otherwise go dead with no forced resync while the store lives.
func TestSQLiteStore_SeqSurvivesEmptyTable(t *testing.T) {
Expand Down
30 changes: 30 additions & 0 deletions internal/timeline/store_conformance_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,30 @@
package timeline

import (
"os"
"path/filepath"
"testing"

"github.com/skyhook-io/radar/pkg/timeline/storetest"
)

// The shared EventStore contract lives in pkg/timeline/storetest and runs here
// against SQLiteStore; MemoryStore runs the same suite from the pkg module.
// Only SQLite-specific behavior (migrations, quarantine, retention, timestamp
// encoding) is tested in this package's own files.
func TestStoreConformance_SQLite(t *testing.T) {
storetest.RunConformance(t, func(t *testing.T) EventStore {
t.Helper()
dir, err := os.MkdirTemp("", "timeline-conformance-*")
if err != nil {
t.Fatalf("MkdirTemp: %v", err)
}
t.Cleanup(func() { os.RemoveAll(dir) })
store, err := NewSQLiteStore(filepath.Join(dir, "test.db"))
if err != nil {
t.Fatalf("NewSQLiteStore: %v", err)
}
t.Cleanup(func() { store.Close() })
return store
})
}
14 changes: 14 additions & 0 deletions pkg/timeline/conformance_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,14 @@
package timeline_test

import (
"testing"

timeline "github.com/skyhook-io/radar/pkg/timeline"
"github.com/skyhook-io/radar/pkg/timeline/storetest"
)

func TestMemoryStore_Conformance(t *testing.T) {
storetest.RunConformance(t, func(t *testing.T) timeline.EventStore {
return timeline.NewMemoryStore(100)
})
}
Loading
Loading