ITWorx Pulse release export
bd774932d5
Public source validation / validate (push) Failing after 3m8s
304 lines
12 KiB
Go
304 lines
12 KiB
Go
package inventory
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"os"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/itworx/pulse/internal/database"
|
|
"github.com/jackc/pgx/v5"
|
|
)
|
|
|
|
func TestInventoryRepositoryPostgreSQL(t *testing.T) {
|
|
dsn := inventoryIntegrationDSN()
|
|
if dsn == "" {
|
|
t.Skip("PULSE_TEST_DATABASE_URL is not set")
|
|
}
|
|
ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second)
|
|
defer cancel()
|
|
pool, err := database.NewPool(ctx, database.Config{URL: dsn, MaxConns: 8, MinConns: 1})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer pool.Close()
|
|
if err := database.Ping(ctx, pool); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := database.Migrate(ctx, pool); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
r, err := NewRepository(pool)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
const sourceA = "00000000-0000-0000-0000-0000000000a1"
|
|
const sourceB = "00000000-0000-0000-0000-0000000000b1"
|
|
const entityID = "00000000-0000-0000-0000-0000000000e1"
|
|
_, _ = pool.Exec(ctx, `DELETE FROM entity_relations WHERE source_id IN ($1,$2); DELETE FROM entity_facts WHERE source_id IN ($1,$2); DELETE FROM entity_aliases WHERE source_id IN ($1,$2); DELETE FROM entities WHERE id=$3; DELETE FROM data_sources WHERE id IN ($1,$2)`, sourceA, sourceB, entityID)
|
|
for _, source := range []string{sourceA, sourceB} {
|
|
if _, err := pool.Exec(ctx, `INSERT INTO data_sources (id,type,name,configuration_ref) VALUES ($1,'exporter',$2,'test') ON CONFLICT (id) DO NOTHING`, source, source); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
now := time.Now().UTC().Truncate(time.Microsecond)
|
|
later := now.Add(time.Second)
|
|
entity := Entity{ID: entityID, EntityType: "container", CanonicalName: "container.test", DisplayName: "Test container", Status: "unknown", FirstSeenAt: &now, LastSeenAt: &now}
|
|
if err := r.PersistDiscovery(ctx, entity, Alias{EntityID: entityID, SourceID: sourceA, ExternalType: "container", ExternalID: "abc"}, []Fact{{EntityID: entityID, FieldName: "image", SourceID: sourceA, Value: []byte(`"one"`), ObservedAt: now, Confidence: 1}}, nil); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
entity.DisplayName = "Updated"
|
|
entity.LastSeenAt = &later
|
|
if err := r.UpsertEntity(ctx, entity); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
got, err := r.GetEntity(ctx, entityID)
|
|
if err != nil || got.DisplayName != "Updated" {
|
|
t.Fatalf("get entity = %+v, err=%v", got, err)
|
|
}
|
|
if err := r.UpsertFact(ctx, Fact{EntityID: entityID, FieldName: "image", SourceID: sourceB, Value: []byte(`"two"`), ObservedAt: now, Confidence: .9}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := r.UpsertOverride(ctx, Override{EntityID: entityID, FieldName: "displayName", Value: []byte(`"Manual"`), UpdatedAt: later}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := r.UpsertRelation(ctx, Relation{ID: "00000000-0000-0000-0000-0000000000f1", SourceEntityID: entityID, TargetEntityID: "00000000-0000-0000-0000-0000000000e2", RelationType: "contains", SourceID: sourceA, Confidence: 1, FirstSeenAt: &now}); err == nil {
|
|
t.Fatal("expected FK failure for missing target entity")
|
|
}
|
|
if err := r.InTx(ctx, func(txctx context.Context, tx pgx.Tx) error {
|
|
if err := upsertFact(txctx, tx, Fact{EntityID: entityID, FieldName: "rollback", SourceID: sourceA, Value: []byte(`true`), ObservedAt: now, Confidence: 1}); err != nil {
|
|
return err
|
|
}
|
|
return errors.New("force rollback")
|
|
}); err == nil {
|
|
t.Fatal("expected transaction callback error")
|
|
}
|
|
count := 0
|
|
if err := pool.QueryRow(ctx, `SELECT count(*) FROM entity_facts WHERE field_name='rollback'`).Scan(&count); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if count != 0 {
|
|
t.Fatal("rolled back fact persisted")
|
|
}
|
|
if err := r.UpsertAlias(ctx, Alias{EntityID: entityID, SourceID: sourceA, ExternalType: "container", ExternalID: "abc"}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := pool.Exec(ctx, `DELETE FROM entities WHERE id=$1`, entityID); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var aliases, facts, overrides int
|
|
if err := pool.QueryRow(ctx, `SELECT (SELECT count(*) FROM entity_aliases WHERE entity_id=$1),(SELECT count(*) FROM entity_facts WHERE entity_id=$1),(SELECT count(*) FROM entity_overrides WHERE entity_id=$1)`, entityID).Scan(&aliases, &facts, &overrides); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if aliases != 0 || facts != 0 || overrides != 0 {
|
|
t.Fatalf("cascade counts aliases=%d facts=%d overrides=%d", aliases, facts, overrides)
|
|
}
|
|
}
|
|
|
|
func TestInventoryRepositoryTargetScaleAndConcurrentUpserts(t *testing.T) {
|
|
dsn := inventoryIntegrationDSN()
|
|
if dsn == "" {
|
|
t.Skip("PULSE_TEST_DATABASE_URL is not set")
|
|
}
|
|
ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second)
|
|
defer cancel()
|
|
pool, err := database.NewPool(ctx, database.Config{URL: dsn, MaxConns: 10, MinConns: 1})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer pool.Close()
|
|
if err := database.Migrate(ctx, pool); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
r, err := NewRepository(pool)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
_, _ = pool.Exec(ctx, `DELETE FROM entities WHERE canonical_name LIKE 'm2-04-scale-%'`)
|
|
now := time.Now().UTC().Truncate(time.Microsecond)
|
|
for i := 1; i <= 490; i++ {
|
|
id := fmt.Sprintf("00000000-0000-0000-0000-%012d", i)
|
|
typeName := "container"
|
|
if i > 150 && i <= 190 {
|
|
typeName = "disk"
|
|
}
|
|
if i > 190 {
|
|
typeName = "probe"
|
|
}
|
|
if err := r.UpsertEntity(ctx, Entity{ID: id, EntityType: typeName, CanonicalName: fmt.Sprintf("m2-04-scale-%04d", i), DisplayName: fmt.Sprintf("Scale %04d", i), Status: "unknown", FirstSeenAt: &now, LastSeenAt: &now}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
var wg sync.WaitGroup
|
|
errs := make(chan error, 10)
|
|
for i := 0; i < 10; i++ {
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
id := "00000000-0000-0000-0000-000000000001"
|
|
for j := 0; j < 10; j++ {
|
|
if err := r.UpsertEntity(ctx, Entity{ID: id, EntityType: "container", CanonicalName: "m2-04-scale-0001", DisplayName: "Concurrent", Status: "unknown", FirstSeenAt: &now, LastSeenAt: &now}); err != nil {
|
|
errs <- err
|
|
return
|
|
}
|
|
}
|
|
}()
|
|
}
|
|
wg.Wait()
|
|
close(errs)
|
|
for err := range errs {
|
|
t.Fatal(err)
|
|
}
|
|
started := time.Now()
|
|
entities, err := r.ListEntities(ctx, 600)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
elapsed := time.Since(started)
|
|
if len(entities) < 490 {
|
|
t.Fatalf("target-scale entity count=%d", len(entities))
|
|
}
|
|
if elapsed >= 500*time.Millisecond {
|
|
t.Fatalf("target-scale inventory list took %s", elapsed)
|
|
}
|
|
_, _ = pool.Exec(ctx, `DELETE FROM entities WHERE canonical_name LIKE 'm2-04-scale-%'`)
|
|
}
|
|
|
|
func TestInventoryReadModelPostgreSQL(t *testing.T) {
|
|
dsn := inventoryIntegrationDSN()
|
|
if dsn == "" {
|
|
t.Skip("PULSE_TEST_DATABASE_URL is not set")
|
|
}
|
|
ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second)
|
|
defer cancel()
|
|
pool, err := database.NewPool(ctx, database.Config{URL: dsn, MaxConns: 8, MinConns: 1})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer pool.Close()
|
|
if err := database.Migrate(ctx, pool); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
r, err := NewRepository(pool)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
const sourceA = "11000000-0000-0000-0000-0000000000a1"
|
|
const sourceB = "11000000-0000-0000-0000-0000000000b1"
|
|
_, _ = pool.Exec(ctx, `DELETE FROM entities WHERE canonical_name LIKE 'm11-05-read-%'`)
|
|
defer func() {
|
|
cleanupCtx, cleanupCancel := context.WithTimeout(context.Background(), 10*time.Second)
|
|
defer cleanupCancel()
|
|
if _, cleanupErr := pool.Exec(cleanupCtx, `DELETE FROM entities WHERE canonical_name LIKE 'm11-05-read-%'`); cleanupErr != nil {
|
|
t.Errorf("clean read-model integration entities: %v", cleanupErr)
|
|
}
|
|
}()
|
|
for _, source := range []string{sourceA, sourceB} {
|
|
if _, err := pool.Exec(ctx, `INSERT INTO data_sources (id,type,name,configuration_ref) VALUES ($1,'exporter',$2,'integration') ON CONFLICT (id) DO UPDATE SET name=EXCLUDED.name`, source, source); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
now := time.Now().UTC().Truncate(time.Microsecond)
|
|
for index := 1; index <= 490; index++ {
|
|
id := fmt.Sprintf("11000000-0000-0000-0000-%012d", index)
|
|
entityType := "container"
|
|
if index > 150 && index <= 190 {
|
|
entityType = "disk"
|
|
} else if index > 190 {
|
|
entityType = "probe"
|
|
}
|
|
if err := r.UpsertEntity(ctx, Entity{ID: id, EntityType: entityType, CanonicalName: fmt.Sprintf("m11-05-read-%03d", index), DisplayName: fmt.Sprintf("Read entity %03d", index), Status: "operational", FirstSeenAt: &now, LastSeenAt: &now}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
const entityID = "11000000-0000-0000-0000-000000000001"
|
|
const peerID = "11000000-0000-0000-0000-000000000002"
|
|
staleAt := now.Add(-time.Minute)
|
|
if err := r.UpsertFact(ctx, Fact{EntityID: entityID, FieldName: "image", SourceID: sourceA, Value: []byte(`"discovered-a"`), ObservedAt: now, Confidence: 1}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := r.UpsertFact(ctx, Fact{EntityID: entityID, FieldName: "image", SourceID: sourceB, Value: []byte(`"discovered-b"`), ObservedAt: now.Add(-time.Hour), Confidence: .8, ValidUntil: &staleAt}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := r.UpsertOverride(ctx, Override{EntityID: entityID, FieldName: "image", Value: []byte(`"manual"`), UpdatedAt: now}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := r.UpsertOverride(ctx, Override{EntityID: entityID, FieldName: "status", Value: []byte(`"attention"`), UpdatedAt: now}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := r.UpsertRelation(ctx, Relation{ID: "11000000-0000-0000-0000-0000000000f1", SourceEntityID: entityID, TargetEntityID: peerID, RelationType: "depends_on", SourceID: sourceA, Confidence: .9, FirstSeenAt: &now}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
seen := map[string]bool{}
|
|
filter := EntityFilter{Limit: 17, Search: "m11-05-read-", Direction: "asc"}
|
|
var slowestPage time.Duration
|
|
for {
|
|
started := time.Now()
|
|
page, err := r.SearchEntities(ctx, filter)
|
|
if elapsed := time.Since(started); elapsed > slowestPage {
|
|
slowestPage = elapsed
|
|
}
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
hasMore := len(page) > filter.Limit
|
|
if hasMore {
|
|
page = page[:filter.Limit]
|
|
}
|
|
for _, item := range page {
|
|
if seen[item.ID] {
|
|
t.Fatalf("duplicate paginated entity %s", item.ID)
|
|
}
|
|
seen[item.ID] = true
|
|
}
|
|
if !hasMore {
|
|
break
|
|
}
|
|
last := page[len(page)-1]
|
|
filter.AfterName, filter.AfterID = strings.ToLower(last.DisplayName), last.ID
|
|
}
|
|
if len(seen) != 490 {
|
|
t.Fatalf("paginated entities=%d want=490", len(seen))
|
|
}
|
|
if slowestPage >= 500*time.Millisecond {
|
|
t.Fatalf("slowest read-model target page took %s", slowestPage)
|
|
}
|
|
disks, err := r.SearchEntities(ctx, EntityFilter{Limit: 100, Search: "m11-05-read-", EntityType: "disk", Direction: "desc"})
|
|
if err != nil || len(disks) != 40 {
|
|
t.Fatalf("disk filter count=%d err=%v", len(disks), err)
|
|
}
|
|
attention, err := r.SearchEntities(ctx, EntityFilter{Limit: 10, Search: "m11-05-read-001", Status: "attention", Direction: "asc"})
|
|
if err != nil || len(attention) != 1 || attention[0].Status != "attention" {
|
|
t.Fatalf("effective status filter=%+v err=%v", attention, err)
|
|
}
|
|
detail, err := r.GetEntityDetail(ctx, entityID)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if len(detail.Facts) != 2 || detail.Entity.SourceCount != 2 || detail.Entity.StaleFactCount != 1 || len(detail.Relations) != 1 {
|
|
t.Fatalf("detail projection=%+v", detail)
|
|
}
|
|
if len(detail.Effective) != 2 || detail.Effective[0].Origin != "override" || string(detail.Effective[0].Value) != `"manual"` {
|
|
t.Fatalf("effective values=%+v", detail.Effective)
|
|
}
|
|
if err := r.PersistDiscovery(ctx, Entity{ID: entityID, EntityType: "container", CanonicalName: "m11-05-read-001", DisplayName: "Discovery rerun", Status: "operational", FirstSeenAt: &now, LastSeenAt: &now}, Alias{EntityID: entityID, SourceID: sourceA, ExternalType: "container", ExternalID: "read-001"}, []Fact{{EntityID: entityID, FieldName: "image", SourceID: sourceA, Value: []byte(`"new-discovery"`), ObservedAt: now.Add(time.Minute), Confidence: 1}}, nil); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
detail, err = r.GetEntityDetail(ctx, entityID)
|
|
if err != nil || detail.Effective[0].Origin != "override" || string(detail.Effective[0].Value) != `"manual"` {
|
|
t.Fatalf("discovery overwrote override: detail=%+v err=%v", detail, err)
|
|
}
|
|
}
|
|
|
|
func inventoryIntegrationDSN() string {
|
|
if dsn := os.Getenv("PULSE_TEST_DATABASE_URL"); dsn != "" {
|
|
return dsn
|
|
}
|
|
return os.Getenv("PULSE_DATABASE_URL")
|
|
}
|