From ecb1ba2bbebc7205d3ab0df2fc87c8d58a96688a Mon Sep 17 00:00:00 2001 From: Jan Eitzinger Date: Thu, 27 Aug 2026 08:06:50 +0200 Subject: [PATCH] feat(fleet): add service type enum and infra scope Replace the free-form service_type string with a ServiceType enum (ccms, ccmc, ccb, cces, ccsa, ccnc, ccem). The short code is the canonical value stored in the DB, the config-tree directory name and a NATS discovery subject token; Description() gives the human-readable name. Add a second scope alongside per-node agents: ScopeInfra covers cluster-independent monitoring infrastructure services, which are identified by (hostname, service_type) and carry an empty cluster. InfraRegistry is the ScopeInfra sibling of Registry and shares the same table, state machine and config-revision handshake. It deliberately exposes no StartSweep: MarkStale ages rows by last_heartbeat regardless of scope, so a single sweep goroutine per process covers both scopes. On the repository side ServiceDB gains a Scope column and the two queries the discovery publisher and infra enumeration need, ListActive() and ListByScope(). ResetConnection() now also resets the fleet repository singleton so tests get a fresh handle. Migration 13 is edited in place rather than superseded by a new migration: the service table is unreleased, so no existing deployment has it yet. Co-Authored-By: Claude Opus 5 (1M context) --- internal/fleet/registry.go | 17 +- internal/fleet/services.go | 160 ++++++++++++++++++ internal/fleet/services_test.go | 127 ++++++++++++++ internal/fleet/servicetype.go | 85 ++++++++++ internal/fleet/servicetype_test.go | 54 ++++++ internal/repository/dbConnection.go | 2 + internal/repository/fleet.go | 66 +++++++- internal/repository/fleet_test.go | 111 ++++++++++++ internal/repository/migration.go | 3 +- .../sqlite3/13_add-service-table.down.sql | 1 + .../sqlite3/13_add-service-table.up.sql | 3 + 11 files changed, 618 insertions(+), 11 deletions(-) create mode 100644 internal/fleet/services.go create mode 100644 internal/fleet/services_test.go create mode 100644 internal/fleet/servicetype.go create mode 100644 internal/fleet/servicetype_test.go create mode 100644 internal/repository/fleet_test.go diff --git a/internal/fleet/registry.go b/internal/fleet/registry.go index 1d683ba2..77319cee 100644 --- a/internal/fleet/registry.go +++ b/internal/fleet/registry.go @@ -21,6 +21,7 @@ import ( "encoding/hex" "encoding/json" "errors" + "fmt" "time" "github.com/ClusterCockpit/cc-backend/internal/repository" @@ -36,7 +37,7 @@ var ErrUnknownInstance = errors.New("fleet: unknown or deregistered instance id" type Service struct { Cluster string Hostname string - ServiceType string + ServiceType ServiceType InstanceID string State string RegisteredAt time.Time @@ -50,7 +51,7 @@ type Service struct { type RegistrationRequest struct { Cluster string Hostname string - ServiceType string + ServiceType ServiceType MetaData map[string]string } @@ -85,8 +86,11 @@ func NewRegistry(staleAfter time.Duration) *Registry { // instance_id plus the config_revision it currently has on record (0 for a // never-before-seen service, so the caller knows to pull its initial config). func (r *Registry) Register(req RegistrationRequest) (*Registration, error) { - if req.Cluster == "" || req.Hostname == "" || req.ServiceType == "" { - return nil, errors.New("fleet: cluster, hostname and service_type are required") + if req.Cluster == "" || req.Hostname == "" { + return nil, errors.New("fleet: cluster and hostname are required") + } + if !req.ServiceType.Valid() { + return nil, fmt.Errorf("fleet: unknown service_type %q", req.ServiceType) } instanceID, err := generateInstanceID() @@ -102,8 +106,9 @@ func (r *Registry) Register(req RegistrationRequest) (*Registration, error) { svc := &repository.ServiceDB{ Cluster: req.Cluster, Hostname: req.Hostname, - ServiceType: req.ServiceType, + ServiceType: string(req.ServiceType), InstanceID: instanceID, + Scope: ScopeCluster, RegisteredAt: time.Now().Unix(), MetaData: metaJSON, } @@ -239,7 +244,7 @@ func toService(row *repository.ServiceDB) (*Service, error) { svc := &Service{ Cluster: row.Cluster, Hostname: row.Hostname, - ServiceType: row.ServiceType, + ServiceType: ServiceType(row.ServiceType), InstanceID: row.InstanceID, State: row.State, RegisteredAt: time.Unix(row.RegisteredAt, 0), diff --git a/internal/fleet/services.go b/internal/fleet/services.go new file mode 100644 index 00000000..cc21f28c --- /dev/null +++ b/internal/fleet/services.go @@ -0,0 +1,160 @@ +// Copyright (C) NHR@FAU, University Erlangen-Nuremberg. +// All rights reserved. This file is part of cc-backend. +// Use of this source code is governed by a MIT-style +// license that can be found in the LICENSE file. + +package fleet + +import ( + "errors" + "fmt" + "time" + + "github.com/ClusterCockpit/cc-backend/internal/repository" + cclog "github.com/ClusterCockpit/cc-lib/v2/ccLogger" +) + +// Scope values distinguish the two kinds of fleet member persisted in the +// shared "service" table: +// +// - ScopeCluster: per-node agents, identified by (cluster, hostname, +// service_type). This is the default and what Registry issues. +// - ScopeInfra: cluster-independent monitoring infrastructure services +// (metric stores, collectors, gateways). They are not tied to a single +// cluster, so their cluster column is empty and identity is +// (hostname, service_type). InfraRegistry issues these. +const ( + ScopeCluster = "cluster" + ScopeInfra = "infra" +) + +// InfraRegistrationRequest is what a cluster-independent monitoring +// infrastructure service posts to the REST registration endpoint. Unlike +// RegistrationRequest it has no Cluster field: these services span clusters. +type InfraRegistrationRequest struct { + Hostname string + ServiceType ServiceType + MetaData map[string]string +} + +// InfraRegistry is the business-logic layer for cluster-independent monitoring +// infrastructure services. It is the ScopeInfra sibling of Registry and shares +// the same FleetRepository and "service" table; identity issuance, the +// pending/active/stale/deregistered state machine and the config-revision +// handshake are identical. The only differences are that registration takes no +// cluster and discovery is by scope rather than by cluster. +// +// Staleness: MarkStale (driven by StartSweep) ages any service by its +// last_heartbeat regardless of scope, so infra rows are already covered by the +// sweep. Do NOT start a second sweep goroutine here — exactly one StartSweep +// across the whole process is sufficient. InfraRegistry therefore deliberately +// exposes no StartSweep. +type InfraRegistry struct { + repo *repository.FleetRepository +} + +// NewInfraRegistry returns an InfraRegistry backed by the singleton +// FleetRepository. +func NewInfraRegistry() *InfraRegistry { + return &InfraRegistry{ + repo: repository.GetFleetRepository(), + } +} + +// Register upserts a cluster-independent service's identity by +// (hostname, service_type) and returns a freshly issued instance_id plus the +// config_revision currently on record (0 for a never-before-seen service, so +// the caller knows to pull its initial config). +func (r *InfraRegistry) Register(req InfraRegistrationRequest) (*Registration, error) { + if req.Hostname == "" { + return nil, errors.New("fleet: hostname is required") + } + if !req.ServiceType.Valid() { + return nil, fmt.Errorf("fleet: unknown service_type %q", req.ServiceType) + } + + instanceID, err := generateInstanceID() + if err != nil { + return nil, err + } + + metaJSON, err := marshalMeta(req.MetaData) + if err != nil { + return nil, err + } + + svc := &repository.ServiceDB{ + Cluster: "", + Hostname: req.Hostname, + ServiceType: string(req.ServiceType), + InstanceID: instanceID, + Scope: ScopeInfra, + RegisteredAt: time.Now().Unix(), + MetaData: metaJSON, + } + + id, err := r.repo.RegisterService(svc) + if err != nil { + return nil, err + } + + stored, err := r.repo.GetByID(id) + if err != nil { + return nil, err + } + + cclog.Infof("fleet: registered infra %s/%s as instance '%s'", req.Hostname, req.ServiceType, instanceID) + return &Registration{InstanceID: instanceID, ConfigRevision: stored.ConfigRevision}, nil +} + +// Heartbeat refreshes liveness for an already-registered infra instance. It is +// a no-op for unknown or deregistered instance IDs — see the package doc for +// why that must hold when this is reachable over NATS. +func (r *InfraRegistry) Heartbeat(instanceID string, at time.Time) error { + affected, err := r.repo.Heartbeat(instanceID, at.Unix()) + if err != nil { + return err + } + if affected == 0 { + return ErrUnknownInstance + } + return nil +} + +// Deregister marks an infra instance as deregistered. Idempotent. +func (r *InfraRegistry) Deregister(instanceID string) error { + return r.repo.Deregister(instanceID) +} + +// AckConfig records that instanceID has pulled configRevision. Called by the +// REST config-pull handler after it serves the config payload. +func (r *InfraRegistry) AckConfig(instanceID string, configRevision int64) error { + return r.repo.SetConfigRevision(instanceID, configRevision) +} + +// Get returns a single infra service by instance_id. +func (r *InfraRegistry) Get(instanceID string) (*Service, error) { + svc, err := r.repo.GetByInstanceID(instanceID) + if err != nil { + return nil, err + } + return toService(svc) +} + +// List returns all registered cluster-independent infrastructure services. +func (r *InfraRegistry) List() ([]*Service, error) { + rows, err := r.repo.ListByScope(ScopeInfra) + if err != nil { + return nil, err + } + + services := make([]*Service, 0, len(rows)) + for _, row := range rows { + svc, err := toService(row) + if err != nil { + return nil, err + } + services = append(services, svc) + } + return services, nil +} diff --git a/internal/fleet/services_test.go b/internal/fleet/services_test.go new file mode 100644 index 00000000..4b8a618f --- /dev/null +++ b/internal/fleet/services_test.go @@ -0,0 +1,127 @@ +// Copyright (C) NHR@FAU, University Erlangen-Nuremberg. +// All rights reserved. This file is part of cc-backend. +// Use of this source code is governed by a MIT-style +// license that can be found in the LICENSE file. + +package fleet + +import ( + "errors" + "path/filepath" + "testing" + "time" + + "github.com/ClusterCockpit/cc-backend/internal/repository" + cclog "github.com/ClusterCockpit/cc-lib/v2/ccLogger" + _ "github.com/mattn/go-sqlite3" +) + +// setupDB migrates a fresh temp database to the current schema and wires the +// repository singletons to it, isolated per test. +func setupDB(t *testing.T) { + t.Helper() + cclog.Init("warn", true) + + dbfile := filepath.Join(t.TempDir(), "fleet.db") + if err := repository.ResetConnection(); err != nil { + t.Fatal(err) + } + if err := repository.MigrateDB(dbfile); err != nil { + t.Fatal(err) + } + repository.Connect(dbfile) + t.Cleanup(func() { repository.ResetConnection() }) +} + +func TestInfraRegistry(t *testing.T) { + setupDB(t) + reg := NewInfraRegistry() + + t.Run("register requires hostname and a valid service_type", func(t *testing.T) { + if _, err := reg.Register(InfraRegistrationRequest{ServiceType: ServiceTypeMetricStore}); err == nil { + t.Fatal("expected error for missing hostname") + } + if _, err := reg.Register(InfraRegistrationRequest{Hostname: "ms01"}); err == nil { + t.Fatal("expected error for missing service_type") + } + if _, err := reg.Register(InfraRegistrationRequest{Hostname: "ms01", ServiceType: "bogus"}); err == nil { + t.Fatal("expected error for unknown service_type") + } + }) + + var instanceID string + t.Run("register issues identity with revision 0", func(t *testing.T) { + r, err := reg.Register(InfraRegistrationRequest{ + Hostname: "ms01", ServiceType: ServiceTypeMetricStore, + MetaData: map[string]string{"version": "1.2.3"}, + }) + if err != nil { + t.Fatal(err) + } + if r.InstanceID == "" || r.ConfigRevision != 0 { + t.Fatalf("unexpected registration: %+v", r) + } + instanceID = r.InstanceID + }) + + t.Run("get returns infra scope and empty cluster", func(t *testing.T) { + svc, err := reg.Get(instanceID) + if err != nil { + t.Fatal(err) + } + if svc.Cluster != "" || svc.ServiceType != ServiceTypeMetricStore || svc.State != "pending" { + t.Fatalf("unexpected service: %+v", svc) + } + if svc.MetaData["version"] != "1.2.3" { + t.Fatalf("metadata not round-tripped: %+v", svc.MetaData) + } + }) + + t.Run("heartbeat activates known, rejects unknown", func(t *testing.T) { + if err := reg.Heartbeat(instanceID, time.Unix(5000, 0)); err != nil { + t.Fatal(err) + } + svc, _ := reg.Get(instanceID) + if svc.State != "active" || svc.LastHeartbeat == nil { + t.Fatalf("heartbeat did not activate: %+v", svc) + } + if err := reg.Heartbeat("unknown", time.Unix(5000, 0)); !errors.Is(err, ErrUnknownInstance) { + t.Fatalf("want ErrUnknownInstance, got %v", err) + } + }) + + t.Run("list returns only infra services", func(t *testing.T) { + // A cluster-scope agent must not appear in the infra listing. + cReg := NewRegistry(time.Hour) + if _, err := cReg.Register(RegistrationRequest{ + Cluster: "fritz", Hostname: "node01", ServiceType: ServiceTypeCollector, + }); err != nil { + t.Fatal(err) + } + + list, err := reg.List() + if err != nil { + t.Fatal(err) + } + if len(list) != 1 || list[0].ServiceType != ServiceTypeMetricStore { + t.Fatalf("unexpected infra list: %+v", list) + } + }) + + t.Run("ack config records revision, deregister is terminal", func(t *testing.T) { + if err := reg.AckConfig(instanceID, 99); err != nil { + t.Fatal(err) + } + svc, _ := reg.Get(instanceID) + if svc.ConfigRevision != 99 { + t.Fatalf("want revision 99, got %d", svc.ConfigRevision) + } + + if err := reg.Deregister(instanceID); err != nil { + t.Fatal(err) + } + if err := reg.Heartbeat(instanceID, time.Unix(6000, 0)); !errors.Is(err, ErrUnknownInstance) { + t.Fatalf("heartbeat after deregister should fail: %v", err) + } + }) +} diff --git a/internal/fleet/servicetype.go b/internal/fleet/servicetype.go new file mode 100644 index 00000000..832a3b6c --- /dev/null +++ b/internal/fleet/servicetype.go @@ -0,0 +1,85 @@ +// Copyright (C) NHR@FAU, University Erlangen-Nuremberg. +// All rights reserved. This file is part of cc-backend. +// Use of this source code is governed by a MIT-style +// license that can be found in the LICENSE file. + +package fleet + +// ServiceType is the kind of cc-* service a fleet member is. It is a +// string-backed enum (mirroring schema.MonitoringState and the local +// ScopeCluster/ScopeInfra constants): the short code is the canonical value +// stored in the DB, used as the config-tree directory name, and used as a NATS +// discovery-subject token. Description() gives the human-readable name. +type ServiceType string + +const ( + ServiceTypeMetricStore ServiceType = "ccms" // cc-metric-store + ServiceTypeCollector ServiceType = "ccmc" // cc-metric-collector + ServiceTypeBackend ServiceType = "ccb" // cc-backend + ServiceTypeEventStore ServiceType = "cces" // cc-event-store + ServiceTypeSlurmAdapter ServiceType = "ccsa" // cc-slurm-adapter + ServiceTypeNodeController ServiceType = "ccnc" // cc-node-controller + ServiceTypeEnergyManager ServiceType = "ccem" // cc-energy-manager +) + +// AllServiceTypes lists every valid service type. Order is stable so callers +// that iterate (e.g. the discovery publisher) produce deterministic output. +var AllServiceTypes = []ServiceType{ + ServiceTypeMetricStore, + ServiceTypeCollector, + ServiceTypeBackend, + ServiceTypeEventStore, + ServiceTypeSlurmAdapter, + ServiceTypeNodeController, + ServiceTypeEnergyManager, +} + +var serviceTypeDescriptions = map[ServiceType]string{ + ServiceTypeMetricStore: "cc-metric-store", + ServiceTypeCollector: "cc-metric-collector", + ServiceTypeBackend: "cc-backend", + ServiceTypeEventStore: "cc-event-store", + ServiceTypeSlurmAdapter: "cc-slurm-adapter", + ServiceTypeNodeController: "cc-node-controller", + ServiceTypeEnergyManager: "cc-energy-manager", +} + +// Valid reports whether t is a known service type. +func (t ServiceType) Valid() bool { + _, ok := serviceTypeDescriptions[t] + return ok +} + +// Description returns the full human-readable service name (e.g. "cc-metric-store"), +// or the raw code if unknown. +func (t ServiceType) Description() string { + if d, ok := serviceTypeDescriptions[t]; ok { + return d + } + return string(t) +} + +// relevantProviders is the single source of truth for which provider service +// types each consumer type needs to discover. Keep it here and edit in one +// place. +// +// Confirmed universal edge: every service must reach cc-backend (ccb) to +// register and pull its config, so ccb is relevant to all of them. Richer +// peer edges (a collector wanting the metric store, an energy manager wanting +// the node controller, …) are left commented for an operator to enable once the +// concrete topology is settled — they are intentionally not assumed here. +var relevantProviders = map[ServiceType][]ServiceType{ + ServiceTypeMetricStore: {ServiceTypeBackend}, + ServiceTypeCollector: {ServiceTypeBackend /*, ServiceTypeMetricStore */}, + ServiceTypeEventStore: {ServiceTypeBackend}, + ServiceTypeSlurmAdapter: {ServiceTypeBackend}, + ServiceTypeNodeController: {ServiceTypeBackend}, + ServiceTypeEnergyManager: {ServiceTypeBackend /*, ServiceTypeMetricStore, ServiceTypeNodeController */}, + ServiceTypeBackend: {}, // ccb discovers no peers by default +} + +// RelevantProviders returns the provider service types that consumer should +// discover. The returned slice must not be mutated by callers. +func RelevantProviders(consumer ServiceType) []ServiceType { + return relevantProviders[consumer] +} diff --git a/internal/fleet/servicetype_test.go b/internal/fleet/servicetype_test.go new file mode 100644 index 00000000..d488802f --- /dev/null +++ b/internal/fleet/servicetype_test.go @@ -0,0 +1,54 @@ +// Copyright (C) NHR@FAU, University Erlangen-Nuremberg. +// All rights reserved. This file is part of cc-backend. +// Use of this source code is governed by a MIT-style +// license that can be found in the LICENSE file. + +package fleet + +import "testing" + +func TestServiceTypeValid(t *testing.T) { + for _, st := range AllServiceTypes { + if !st.Valid() { + t.Errorf("%q should be valid", st) + } + } + for _, bad := range []ServiceType{"", "bogus", "metric-store", "CCMS"} { + if bad.Valid() { + t.Errorf("%q should be invalid", bad) + } + } +} + +func TestServiceTypeDescription(t *testing.T) { + if got := ServiceTypeMetricStore.Description(); got != "cc-metric-store" { + t.Errorf("ccms description = %q, want cc-metric-store", got) + } + // Unknown code falls back to the raw string. + if got := ServiceType("bogus").Description(); got != "bogus" { + t.Errorf("unknown description = %q, want bogus", got) + } +} + +func TestRelevantProviders(t *testing.T) { + // Confirmed universal edge: every non-ccb service must discover ccb. + for _, st := range AllServiceTypes { + if st == ServiceTypeBackend { + continue + } + rel := RelevantProviders(st) + found := false + for _, p := range rel { + if p == ServiceTypeBackend { + found = true + } + } + if !found { + t.Errorf("%q must have ccb as a relevant provider, got %v", st, rel) + } + } + // ccb discovers no peers by default. + if rel := RelevantProviders(ServiceTypeBackend); len(rel) != 0 { + t.Errorf("ccb should have no relevant providers, got %v", rel) + } +} diff --git a/internal/repository/dbConnection.go b/internal/repository/dbConnection.go index 80df2869..9cd8c3f2 100644 --- a/internal/repository/dbConnection.go +++ b/internal/repository/dbConnection.go @@ -136,6 +136,8 @@ func ResetConnection() error { userRepoOnce = sync.Once{} userCfgRepoInstance = nil userCfgRepoOnce = sync.Once{} + fleetRepoInstance = nil + fleetRepoOnce = sync.Once{} return nil } diff --git a/internal/repository/fleet.go b/internal/repository/fleet.go index 6393db4d..ec479fc2 100644 --- a/internal/repository/fleet.go +++ b/internal/repository/fleet.go @@ -52,11 +52,15 @@ type ServiceDB struct { LastHeartbeat sql.NullInt64 `db:"last_heartbeat"` ConfigRevision int64 `db:"config_revision"` MetaData sql.NullString `db:"meta_data"` + // Scope is 'cluster' for per-node agents (the default) or 'infra' for + // cluster-independent monitoring infrastructure services. Infra rows carry + // an empty cluster. + Scope string `db:"scope"` } const namedServiceInsert string = ` -INSERT INTO service (cluster, hostname, service_type, instance_id, state, registered_at, config_revision, meta_data) - VALUES (:cluster, :hostname, :service_type, :instance_id, 'pending', :registered_at, 0, :meta_data);` +INSERT INTO service (cluster, hostname, service_type, instance_id, scope, state, registered_at, config_revision, meta_data) + VALUES (:cluster, :hostname, :service_type, :instance_id, :scope, 'pending', :registered_at, 0, :meta_data);` // RegisterService upserts a service by (cluster, hostname, service_type): a // service registering for the first time is inserted with config_revision 0; @@ -75,6 +79,7 @@ func (r *FleetRepository) RegisterService(svc *ServiceDB) (int64, error) { case nil: if _, uerr := sq.Update("service"). Set("instance_id", svc.InstanceID). + Set("scope", svc.Scope). Set("state", "pending"). Set("registered_at", svc.RegisteredAt). Set("meta_data", svc.MetaData). @@ -156,13 +161,13 @@ func (r *FleetRepository) SetConfigRevision(instanceID string, revision int64) e var serviceColumns = []string{ "id", "cluster", "hostname", "service_type", "instance_id", - "state", "registered_at", "last_heartbeat", "config_revision", "meta_data", + "state", "registered_at", "last_heartbeat", "config_revision", "meta_data", "scope", } func scanService(row interface{ Scan(...any) error }) (*ServiceDB, error) { svc := &ServiceDB{} if err := row.Scan(&svc.ID, &svc.Cluster, &svc.Hostname, &svc.ServiceType, &svc.InstanceID, - &svc.State, &svc.RegisteredAt, &svc.LastHeartbeat, &svc.ConfigRevision, &svc.MetaData); err != nil { + &svc.State, &svc.RegisteredAt, &svc.LastHeartbeat, &svc.ConfigRevision, &svc.MetaData, &svc.Scope); err != nil { return nil, err } return svc, nil @@ -210,3 +215,56 @@ func (r *FleetRepository) ListByCluster(cluster string) ([]*ServiceDB, error) { } return services, rows.Err() } + +// ListActive returns all services currently in the 'active' state, ordered by +// cluster then hostname. It is the roster source for the fleet discovery +// publisher: only live services are advertised to peers. +func (r *FleetRepository) ListActive() ([]*ServiceDB, error) { + rows, err := sq.Select(serviceColumns...).From("service"). + Where("state = ?", "active"). + OrderBy("cluster ASC", "hostname ASC"). + RunWith(r.DB).Query() + if err != nil { + cclog.Errorf("Error while listing active services: %v", err) + return nil, err + } + defer rows.Close() + + services := make([]*ServiceDB, 0) + for rows.Next() { + svc, err := scanService(rows) + if err != nil { + cclog.Warn("Error while scanning rows (ListActive)") + return nil, err + } + services = append(services, svc) + } + return services, rows.Err() +} + +// ListByScope returns all services with the given scope ('cluster' or +// 'infra'), ordered by hostname/service_type. Used to enumerate +// cluster-independent monitoring infrastructure services, which carry an empty +// cluster and therefore are not discoverable via ListByCluster. +func (r *FleetRepository) ListByScope(scope string) ([]*ServiceDB, error) { + rows, err := sq.Select(serviceColumns...).From("service"). + Where("scope = ?", scope). + OrderBy("hostname ASC", "service_type ASC"). + RunWith(r.DB).Query() + if err != nil { + cclog.Errorf("Error while listing services for scope '%s': %v", scope, err) + return nil, err + } + defer rows.Close() + + services := make([]*ServiceDB, 0) + for rows.Next() { + svc, err := scanService(rows) + if err != nil { + cclog.Warn("Error while scanning rows (ListByScope)") + return nil, err + } + services = append(services, svc) + } + return services, rows.Err() +} diff --git a/internal/repository/fleet_test.go b/internal/repository/fleet_test.go new file mode 100644 index 00000000..2b106d3d --- /dev/null +++ b/internal/repository/fleet_test.go @@ -0,0 +1,111 @@ +// Copyright (C) NHR@FAU, University Erlangen-Nuremberg. +// All rights reserved. This file is part of cc-backend. +// Use of this source code is governed by a MIT-style +// license that can be found in the LICENSE file. +package repository + +import ( + "testing" +) + +func TestFleetRepository(t *testing.T) { + setup(t) // migrates a fresh temp DB to the current schema version + repo := GetFleetRepository() + + // A per-node (cluster) agent and a cluster-independent infra service. + clusterSvc := &ServiceDB{ + Cluster: "fritz", Hostname: "node01", ServiceType: "agent", + InstanceID: "iid-cluster", Scope: "cluster", RegisteredAt: 1000, + } + infraSvc := &ServiceDB{ + Cluster: "", Hostname: "ms01", ServiceType: "metric-store", + InstanceID: "iid-infra", Scope: "infra", RegisteredAt: 2000, + } + + if _, err := repo.RegisterService(clusterSvc); err != nil { + t.Fatal(err) + } + if _, err := repo.RegisterService(infraSvc); err != nil { + t.Fatal(err) + } + + t.Run("ListByScope isolates infra from cluster", func(t *testing.T) { + infra, err := repo.ListByScope("infra") + noErr(t, err) + if len(infra) != 1 { + t.Fatalf("want 1 infra service, got %d", len(infra)) + } + if infra[0].InstanceID != "iid-infra" || infra[0].Cluster != "" || infra[0].Scope != "infra" { + t.Fatalf("unexpected infra row: %+v", infra[0]) + } + + cluster, err := repo.ListByScope("cluster") + noErr(t, err) + if len(cluster) != 1 || cluster[0].InstanceID != "iid-cluster" { + t.Fatalf("unexpected cluster scope rows: %+v", cluster) + } + }) + + t.Run("ListByCluster sees only its cluster", func(t *testing.T) { + rows, err := repo.ListByCluster("fritz") + noErr(t, err) + if len(rows) != 1 || rows[0].InstanceID != "iid-cluster" { + t.Fatalf("unexpected ListByCluster(fritz): %+v", rows) + } + }) + + t.Run("scope round-trips through GetByInstanceID", func(t *testing.T) { + got, err := repo.GetByInstanceID("iid-infra") + noErr(t, err) + if got.Scope != "infra" || got.State != "pending" { + t.Fatalf("unexpected: scope=%q state=%q", got.Scope, got.State) + } + }) + + t.Run("heartbeat is instance-id keyed and scope-agnostic", func(t *testing.T) { + affected, err := repo.Heartbeat("iid-infra", 3000) + noErr(t, err) + if affected != 1 { + t.Fatalf("want 1 row affected, got %d", affected) + } + affected, err = repo.Heartbeat("does-not-exist", 3000) + noErr(t, err) + if affected != 0 { + t.Fatalf("want 0 rows affected for unknown instance, got %d", affected) + } + }) + + t.Run("ListActive returns only active services", func(t *testing.T) { + // iid-infra was activated by the heartbeat above; iid-cluster is still pending. + active, err := repo.ListActive() + noErr(t, err) + if len(active) != 1 || active[0].InstanceID != "iid-infra" || active[0].State != "active" { + t.Fatalf("unexpected ListActive result: %+v", active) + } + }) + + t.Run("config revision persists", func(t *testing.T) { + noErr(t, repo.SetConfigRevision("iid-infra", 42)) + got, err := repo.GetByInstanceID("iid-infra") + noErr(t, err) + if got.ConfigRevision != 42 { + t.Fatalf("want config_revision 42, got %d", got.ConfigRevision) + } + }) + + t.Run("re-register keeps revision, resets scope and state", func(t *testing.T) { + // Same identity triple, but arriving as an infra registration again. + reReg := &ServiceDB{ + Cluster: "", Hostname: "ms01", ServiceType: "metric-store", + InstanceID: "iid-infra-2", Scope: "infra", RegisteredAt: 4000, + } + if _, err := repo.RegisterService(reReg); err != nil { + t.Fatal(err) + } + got, err := repo.GetByInstanceID("iid-infra-2") + noErr(t, err) + if got.State != "pending" || got.Scope != "infra" || got.ConfigRevision != 42 { + t.Fatalf("unexpected after re-register: %+v", got) + } + }) +} diff --git a/internal/repository/migration.go b/internal/repository/migration.go index 8d2e1460..1c124dde 100644 --- a/internal/repository/migration.go +++ b/internal/repository/migration.go @@ -21,7 +21,8 @@ import ( // is added to internal/repository/migrations/sqlite3/. // // Version history: -// - Version 13: Service table (fleet service registration/heartbeat) +// - Version 13: Service table (fleet service registration/heartbeat), incl. +// scope column (cluster vs cluster-independent infra services) // - Version 12: Sessions table (server-side sessions via alexedwards/scs) // - Version 11: Optimize job table indexes (reduce from ~78 to 48, add covering/partial indexes) // - Version 10: Node table diff --git a/internal/repository/migrations/sqlite3/13_add-service-table.down.sql b/internal/repository/migrations/sqlite3/13_add-service-table.down.sql index 755e00bb..e439c173 100644 --- a/internal/repository/migrations/sqlite3/13_add-service-table.down.sql +++ b/internal/repository/migrations/sqlite3/13_add-service-table.down.sql @@ -1,3 +1,4 @@ DROP INDEX IF EXISTS services_state_heartbeat; +DROP INDEX IF EXISTS services_scope; DROP INDEX IF EXISTS services_cluster; DROP TABLE IF EXISTS "service"; diff --git a/internal/repository/migrations/sqlite3/13_add-service-table.up.sql b/internal/repository/migrations/sqlite3/13_add-service-table.up.sql index d23fe56f..53d7bdab 100644 --- a/internal/repository/migrations/sqlite3/13_add-service-table.up.sql +++ b/internal/repository/migrations/sqlite3/13_add-service-table.up.sql @@ -4,6 +4,8 @@ CREATE TABLE "service" ( hostname VARCHAR(255) NOT NULL, service_type VARCHAR(255) NOT NULL, instance_id VARCHAR(64) NOT NULL, + scope VARCHAR(32) NOT NULL DEFAULT 'cluster' + CHECK (scope IN ('cluster', 'infra')), state VARCHAR(32) NOT NULL DEFAULT 'pending' CHECK (state IN ('pending', 'active', 'stale', 'deregistered')), registered_at INTEGER NOT NULL, @@ -15,6 +17,7 @@ CREATE TABLE "service" ( ); CREATE INDEX IF NOT EXISTS services_cluster ON service (cluster); +CREATE INDEX IF NOT EXISTS services_scope ON service (scope); CREATE INDEX IF NOT EXISTS services_state_heartbeat ON service (state, last_heartbeat); PRAGMA optimize;