ITWorx Pulse release export
bd774932d5
Public source validation / validate (push) Failing after 3m8s
106 lines
4.2 KiB
Go
106 lines
4.2 KiB
Go
package workerruntime
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"os"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/itworx/pulse/internal/container"
|
|
"github.com/itworx/pulse/internal/database"
|
|
"github.com/itworx/pulse/internal/discovery"
|
|
"github.com/itworx/pulse/internal/inventory"
|
|
)
|
|
|
|
// TestPostgreSQLDiscoveryPersistsInventoryProvenanceAtTargetScale proves the
|
|
// complete worker-to-inventory persistence path against real PostgreSQL. The
|
|
// v1 target of 150 active containers is replayed in a later observation window
|
|
// to verify stable application IDs, fact upserts and relation uniqueness.
|
|
func TestPostgreSQLDiscoveryPersistsInventoryProvenanceAtTargetScale(t *testing.T) {
|
|
dsn := os.Getenv("PULSE_TEST_DATABASE_URL")
|
|
if dsn == "" {
|
|
t.Skip("PULSE_TEST_DATABASE_URL is not set")
|
|
}
|
|
ctx, cancel := context.WithTimeout(context.Background(), 120*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)
|
|
}
|
|
sourceID := newID()
|
|
if _, err := pool.Exec(ctx, `INSERT INTO data_sources (id,type,name,configuration_ref) VALUES ($1,'agent',$2,'integration')`, sourceID, "m12-discovery-"+sourceID); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(func() {
|
|
cleanupCtx, cleanupCancel := context.WithTimeout(context.Background(), 30*time.Second)
|
|
defer cleanupCancel()
|
|
_, _ = pool.Exec(cleanupCtx, `DELETE FROM entity_relations WHERE source_id=$1`, sourceID)
|
|
_, _ = pool.Exec(cleanupCtx, `DELETE FROM entity_facts WHERE source_id=$1`, sourceID)
|
|
_, _ = pool.Exec(cleanupCtx, `DELETE FROM container_aliases WHERE source_id=$1`, sourceID)
|
|
_, _ = pool.Exec(cleanupCtx, `DELETE FROM entities WHERE id IN (SELECT entity_id FROM entity_aliases WHERE source_id=$1)`, sourceID)
|
|
_, _ = pool.Exec(cleanupCtx, `DELETE FROM data_sources WHERE id=$1`, sourceID)
|
|
})
|
|
|
|
repository, err := inventory.NewRepository(pool)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
now := time.Now().UTC().Truncate(time.Second)
|
|
items := make([]container.Container, 0, 150)
|
|
for index := 0; index < 150; index++ {
|
|
name := fmt.Sprintf("m12-scale-%03d", index)
|
|
items = append(items, container.Container{ID: "runtime-" + name, Name: name, State: "running", Health: "healthy", MetricsAvailable: true, LifecycleAvailable: true})
|
|
}
|
|
provider := &fakeContainerProvider{snapshot: healthySnapshot(now, items...)}
|
|
job := DiscoveryJob{
|
|
SourceID: sourceID, Provider: provider, Aliases: PostgresContainerAliasStore{Pool: pool}, Inventory: repository,
|
|
Runner: discovery.Runner{Store: discovery.NewMemoryStore(), MaxAttempts: 1},
|
|
}
|
|
if _, err := job.Run(ctx); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
assertInventoryScale := func() {
|
|
t.Helper()
|
|
var entities, facts, relations int
|
|
if err := pool.QueryRow(ctx, `SELECT count(DISTINCT entity_id) FROM entity_aliases WHERE source_id=$1`, sourceID).Scan(&entities); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := pool.QueryRow(ctx, `SELECT count(*) FROM entity_facts WHERE source_id=$1`, sourceID).Scan(&facts); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := pool.QueryRow(ctx, `SELECT count(*) FROM entity_relations WHERE source_id=$1`, sourceID).Scan(&relations); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if entities != 300 || facts < 1_200 || relations != 150 {
|
|
t.Fatalf("inventory scale entities=%d facts=%d relations=%d", entities, facts, relations)
|
|
}
|
|
}
|
|
assertInventoryScale()
|
|
|
|
provider.set(healthySnapshot(now.Add(time.Minute), items...))
|
|
if _, err := job.Run(ctx); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
assertInventoryScale()
|
|
|
|
provider.set(healthySnapshot(now.Add(2 * time.Minute)))
|
|
if _, err := job.Run(ctx); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var activeEntities, tombstonedRelations int
|
|
if err := pool.QueryRow(ctx, `SELECT count(*) FROM entities e JOIN entity_aliases a ON a.entity_id=e.id WHERE a.source_id=$1 AND e.tombstoned_at IS NULL`, sourceID).Scan(&activeEntities); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := pool.QueryRow(ctx, `SELECT count(*) FROM entity_relations WHERE source_id=$1 AND tombstoned_at IS NOT NULL`, sourceID).Scan(&tombstonedRelations); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if activeEntities != 0 || tombstonedRelations != 150 {
|
|
t.Fatalf("disappearance reconciliation active=%d tombstonedRelations=%d", activeEntities, tombstonedRelations)
|
|
}
|
|
}
|