mirror of
https://github.com/ClusterCockpit/cc-backend
synced 2026-08-31 08:57:14 +02:00
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) <noreply@anthropic.com>
This commit is contained in:
@@ -136,6 +136,8 @@ func ResetConnection() error {
|
||||
userRepoOnce = sync.Once{}
|
||||
userCfgRepoInstance = nil
|
||||
userCfgRepoOnce = sync.Once{}
|
||||
fleetRepoInstance = nil
|
||||
fleetRepoOnce = sync.Once{}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -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()
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
})
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -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";
|
||||
|
||||
@@ -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;
|
||||
|
||||
Reference in New Issue
Block a user