From 5f94ae3b68c76e7539faa9287ce0f4513d6dd034 Mon Sep 17 00:00:00 2001 From: Jan Eitzinger Date: Fri, 24 Jul 2026 07:02:09 +0200 Subject: [PATCH] Initial template for registration service --- internal/fleet/registry.go | 254 ++++++++++++++++++ internal/repository/fleet.go | 212 +++++++++++++++ internal/repository/migration.go | 3 +- .../sqlite3/13_add-service-table.down.sql | 3 + .../sqlite3/13_add-service-table.up.sql | 20 ++ 5 files changed, 491 insertions(+), 1 deletion(-) create mode 100644 internal/fleet/registry.go create mode 100644 internal/repository/fleet.go create mode 100644 internal/repository/migrations/sqlite3/13_add-service-table.down.sql create mode 100644 internal/repository/migrations/sqlite3/13_add-service-table.up.sql diff --git a/internal/fleet/registry.go b/internal/fleet/registry.go new file mode 100644 index 00000000..1d683ba2 --- /dev/null +++ b/internal/fleet/registry.go @@ -0,0 +1,254 @@ +// 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 provides central service discovery and configuration +// deployment for auxiliary cc-* services (e.g. metric collectors, metric +// stores) running across a cluster. +// +// Registration and config pull happen over authenticated REST — that is the +// only path that can create or resurrect a service identity. NATS, which has +// no application-layer auth (see CLAUDE.md), may only be used to carry +// heartbeats for an instance_id that REST already issued; an unknown or +// deregistered instance_id is a no-op, never an upsert. +package fleet + +import ( + "context" + "crypto/rand" + "database/sql" + "encoding/hex" + "encoding/json" + "errors" + "time" + + "github.com/ClusterCockpit/cc-backend/internal/repository" + cclog "github.com/ClusterCockpit/cc-lib/v2/ccLogger" +) + +// ErrUnknownInstance is returned by Heartbeat for an instance_id that was +// never registered (or has been deregistered) — it is never auto-created. +var ErrUnknownInstance = errors.New("fleet: unknown or deregistered instance id") + +// Service is the domain-facing view of a fleet registration, as returned to +// REST handlers/GraphQL resolvers. +type Service struct { + Cluster string + Hostname string + ServiceType string + InstanceID string + State string + RegisteredAt time.Time + LastHeartbeat *time.Time + ConfigRevision int64 + MetaData map[string]string +} + +// RegistrationRequest is what a service posts to the REST registration +// endpoint. +type RegistrationRequest struct { + Cluster string + Hostname string + ServiceType string + MetaData map[string]string +} + +// Registration is returned from Register: the instance_id must be presented +// on every subsequent heartbeat and config pull. +type Registration struct { + InstanceID string + ConfigRevision int64 +} + +// Registry is the business-logic layer on top of FleetRepository: it owns +// instance-identity issuance and the pending/active/stale/deregistered state +// machine. REST handlers call Register/Deregister/List/Get; a NATS heartbeat +// consumer (mirroring the worker-pool shape in internal/api/nats.go) should +// call only Heartbeat. +type Registry struct { + repo *repository.FleetRepository + staleAfter time.Duration +} + +// NewRegistry returns a Registry backed by the singleton FleetRepository. +// staleAfter is how long a service may go without a heartbeat before StartSweep +// flips it from 'active' to 'stale'. +func NewRegistry(staleAfter time.Duration) *Registry { + return &Registry{ + repo: repository.GetFleetRepository(), + staleAfter: staleAfter, + } +} + +// Register upserts a service's identity and returns a freshly issued +// 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") + } + + instanceID, err := generateInstanceID() + if err != nil { + return nil, err + } + + metaJSON, err := marshalMeta(req.MetaData) + if err != nil { + return nil, err + } + + svc := &repository.ServiceDB{ + Cluster: req.Cluster, + Hostname: req.Hostname, + ServiceType: req.ServiceType, + InstanceID: instanceID, + 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 %s/%s/%s as instance '%s'", req.Cluster, req.Hostname, req.ServiceType, instanceID) + return &Registration{InstanceID: instanceID, ConfigRevision: stored.ConfigRevision}, nil +} + +// Heartbeat refreshes liveness for an already-registered 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 *Registry) 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 instance as deregistered. Idempotent. +func (r *Registry) 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 *Registry) AckConfig(instanceID string, configRevision int64) error { + return r.repo.SetConfigRevision(instanceID, configRevision) +} + +// Get returns a single service by instance_id. +func (r *Registry) Get(instanceID string) (*Service, error) { + svc, err := r.repo.GetByInstanceID(instanceID) + if err != nil { + return nil, err + } + return toService(svc) +} + +// List returns all services registered for a cluster. +func (r *Registry) List(cluster string) ([]*Service, error) { + rows, err := r.repo.ListByCluster(cluster) + 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 +} + +// StartSweep runs until ctx is cancelled, periodically marking services whose +// last heartbeat is older than staleAfter as 'stale'. Mirrors the worker +// goroutine lifecycle already used for NATS job/node consumers. +func (r *Registry) StartSweep(ctx context.Context, interval time.Duration) { + go func() { + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + cutoff := time.Now().Add(-r.staleAfter).Unix() + n, err := r.repo.MarkStale(cutoff) + if err != nil { + cclog.Errorf("fleet: stale sweep failed: %v", err) + continue + } + if n > 0 { + cclog.Infof("fleet: marked %d service(s) stale", n) + } + } + } + }() +} + +func generateInstanceID() (string, error) { + buf := make([]byte, 16) + if _, err := rand.Read(buf); err != nil { + return "", err + } + return hex.EncodeToString(buf), nil +} + +func marshalMeta(meta map[string]string) (sql.NullString, error) { + if len(meta) == 0 { + return sql.NullString{}, nil + } + buf, err := json.Marshal(meta) + if err != nil { + return sql.NullString{}, err + } + return sql.NullString{String: string(buf), Valid: true}, nil +} + +func unmarshalMeta(raw sql.NullString) (map[string]string, error) { + if !raw.Valid || raw.String == "" { + return nil, nil + } + meta := make(map[string]string) + if err := json.Unmarshal([]byte(raw.String), &meta); err != nil { + return nil, err + } + return meta, nil +} + +func toService(row *repository.ServiceDB) (*Service, error) { + meta, err := unmarshalMeta(row.MetaData) + if err != nil { + return nil, err + } + + svc := &Service{ + Cluster: row.Cluster, + Hostname: row.Hostname, + ServiceType: row.ServiceType, + InstanceID: row.InstanceID, + State: row.State, + RegisteredAt: time.Unix(row.RegisteredAt, 0), + ConfigRevision: row.ConfigRevision, + MetaData: meta, + } + if row.LastHeartbeat.Valid { + t := time.Unix(row.LastHeartbeat.Int64, 0) + svc.LastHeartbeat = &t + } + return svc, nil +} diff --git a/internal/repository/fleet.go b/internal/repository/fleet.go new file mode 100644 index 00000000..6393db4d --- /dev/null +++ b/internal/repository/fleet.go @@ -0,0 +1,212 @@ +// 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 ( + "database/sql" + "sync" + + cclog "github.com/ClusterCockpit/cc-lib/v2/ccLogger" + "github.com/ClusterCockpit/cc-lib/v2/lrucache" + sq "github.com/Masterminds/squirrel" + "github.com/jmoiron/sqlx" +) + +var ( + fleetRepoOnce sync.Once + fleetRepoInstance *FleetRepository +) + +type FleetRepository struct { + DB *sqlx.DB + cache *lrucache.Cache + driver string +} + +func GetFleetRepository() *FleetRepository { + fleetRepoOnce.Do(func() { + db := GetConnection() + + fleetRepoInstance = &FleetRepository{ + DB: db.DB, + driver: db.Driver, + cache: lrucache.New(repoConfig.CacheSize), + } + }) + return fleetRepoInstance +} + +// ServiceDB is the row shape of the "service" table: one row per registered +// (cluster, hostname, service_type) fleet member. +type ServiceDB struct { + ID int64 `db:"id"` + Cluster string `db:"cluster"` + Hostname string `db:"hostname"` + ServiceType string `db:"service_type"` + InstanceID string `db:"instance_id"` + State string `db:"state"` + RegisteredAt int64 `db:"registered_at"` + LastHeartbeat sql.NullInt64 `db:"last_heartbeat"` + ConfigRevision int64 `db:"config_revision"` + MetaData sql.NullString `db:"meta_data"` +} + +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);` + +// RegisterService upserts a service by (cluster, hostname, service_type): a +// service registering for the first time is inserted with config_revision 0; +// one re-registering (e.g. after a restart) keeps its existing last_heartbeat +// and config_revision but gets a fresh instance_id and is reset to 'pending' +// until its next heartbeat. Returns the row id. +func (r *FleetRepository) RegisterService(svc *ServiceDB) (int64, error) { + var id int64 + err := sq.Select("id").From("service"). + Where("cluster = ?", svc.Cluster). + Where("hostname = ?", svc.Hostname). + Where("service_type = ?", svc.ServiceType). + RunWith(r.DB).QueryRow().Scan(&id) + + switch err { + case nil: + if _, uerr := sq.Update("service"). + Set("instance_id", svc.InstanceID). + Set("state", "pending"). + Set("registered_at", svc.RegisteredAt). + Set("meta_data", svc.MetaData). + Where("id = ?", id). + RunWith(r.DB).Exec(); uerr != nil { + cclog.Errorf("Error while re-registering service '%s/%s/%s': %v", svc.Cluster, svc.Hostname, svc.ServiceType, uerr) + return 0, uerr + } + return id, nil + case sql.ErrNoRows: + res, ierr := r.DB.NamedExec(namedServiceInsert, svc) + if ierr != nil { + cclog.Errorf("Error while registering service '%s/%s/%s': %v", svc.Cluster, svc.Hostname, svc.ServiceType, ierr) + return 0, ierr + } + return res.LastInsertId() + default: + cclog.Errorf("Error while looking up service '%s/%s/%s': %v", svc.Cluster, svc.Hostname, svc.ServiceType, err) + return 0, err + } +} + +// Heartbeat marks the instance active as of timestamp. It never creates a row: +// the returned row count is 0 for an unknown or deregistered instance_id, so +// callers (e.g. an unauthenticated NATS consumer) can tell a spoofed/expired +// heartbeat from a real one without the repository silently upserting it. +func (r *FleetRepository) Heartbeat(instanceID string, timestamp int64) (int64, error) { + res, err := sq.Update("service"). + Set("last_heartbeat", timestamp). + Set("state", "active"). + Where("instance_id = ?", instanceID). + Where("state <> ?", "deregistered"). + RunWith(r.DB).Exec() + if err != nil { + cclog.Errorf("Error while recording heartbeat for instance '%s': %v", instanceID, err) + return 0, err + } + return res.RowsAffected() +} + +// MarkStale flips 'active' services with last_heartbeat older than cutoff to +// 'stale'. Intended to be called periodically by a sweep goroutine. +func (r *FleetRepository) MarkStale(cutoff int64) (int64, error) { + res, err := sq.Update("service"). + Set("state", "stale"). + Where("state = ?", "active"). + Where("last_heartbeat < ?", cutoff). + RunWith(r.DB).Exec() + if err != nil { + cclog.Errorf("Error while marking stale services (cutoff %d): %v", cutoff, err) + return 0, err + } + return res.RowsAffected() +} + +func (r *FleetRepository) Deregister(instanceID string) error { + if _, err := sq.Update("service"). + Set("state", "deregistered"). + Where("instance_id = ?", instanceID). + RunWith(r.DB).Exec(); err != nil { + cclog.Errorf("Error while deregistering instance '%s': %v", instanceID, err) + return err + } + return nil +} + +// SetConfigRevision records which config revision a service last received. +// Called after the service pulls its config over REST. +func (r *FleetRepository) SetConfigRevision(instanceID string, revision int64) error { + if _, err := sq.Update("service"). + Set("config_revision", revision). + Where("instance_id = ?", instanceID). + RunWith(r.DB).Exec(); err != nil { + cclog.Errorf("Error while setting config revision for instance '%s': %v", instanceID, err) + return err + } + return nil +} + +var serviceColumns = []string{ + "id", "cluster", "hostname", "service_type", "instance_id", + "state", "registered_at", "last_heartbeat", "config_revision", "meta_data", +} + +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 { + return nil, err + } + return svc, nil +} + +func (r *FleetRepository) GetByID(id int64) (*ServiceDB, error) { + row := sq.Select(serviceColumns...).From("service").Where("id = ?", id).RunWith(r.DB).QueryRow() + svc, err := scanService(row) + if err != nil { + cclog.Errorf("Error while querying service id '%d': %v", id, err) + return nil, err + } + return svc, nil +} + +func (r *FleetRepository) GetByInstanceID(instanceID string) (*ServiceDB, error) { + row := sq.Select(serviceColumns...).From("service").Where("instance_id = ?", instanceID).RunWith(r.DB).QueryRow() + svc, err := scanService(row) + if err != nil { + cclog.Errorf("Error while querying service instance '%s': %v", instanceID, err) + return nil, err + } + return svc, nil +} + +func (r *FleetRepository) ListByCluster(cluster string) ([]*ServiceDB, error) { + rows, err := sq.Select(serviceColumns...).From("service"). + Where("cluster = ?", cluster). + OrderBy("hostname ASC", "service_type ASC"). + RunWith(r.DB).Query() + if err != nil { + cclog.Errorf("Error while listing services for cluster '%s': %v", cluster, 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 (ListByCluster)") + return nil, err + } + services = append(services, svc) + } + return services, rows.Err() +} diff --git a/internal/repository/migration.go b/internal/repository/migration.go index b97db97d..8d2e1460 100644 --- a/internal/repository/migration.go +++ b/internal/repository/migration.go @@ -21,12 +21,13 @@ import ( // is added to internal/repository/migrations/sqlite3/. // // Version history: +// - Version 13: Service table (fleet service registration/heartbeat) // - 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 // // Migration files are embedded at build time from the migrations directory. -const Version uint = 12 +const Version uint = 13 //go:embed migrations/* var migrationFiles embed.FS diff --git a/internal/repository/migrations/sqlite3/13_add-service-table.down.sql b/internal/repository/migrations/sqlite3/13_add-service-table.down.sql new file mode 100644 index 00000000..755e00bb --- /dev/null +++ b/internal/repository/migrations/sqlite3/13_add-service-table.down.sql @@ -0,0 +1,3 @@ +DROP INDEX IF EXISTS services_state_heartbeat; +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 new file mode 100644 index 00000000..d23fe56f --- /dev/null +++ b/internal/repository/migrations/sqlite3/13_add-service-table.up.sql @@ -0,0 +1,20 @@ +CREATE TABLE "service" ( + id INTEGER PRIMARY KEY, + cluster VARCHAR(255) NOT NULL, + hostname VARCHAR(255) NOT NULL, + service_type VARCHAR(255) NOT NULL, + instance_id VARCHAR(64) NOT NULL, + state VARCHAR(32) NOT NULL DEFAULT 'pending' + CHECK (state IN ('pending', 'active', 'stale', 'deregistered')), + registered_at INTEGER NOT NULL, + last_heartbeat INTEGER, + config_revision INTEGER NOT NULL DEFAULT 0, + meta_data TEXT, -- JSON + UNIQUE (cluster, hostname, service_type), + UNIQUE (instance_id) +); + +CREATE INDEX IF NOT EXISTS services_cluster ON service (cluster); +CREATE INDEX IF NOT EXISTS services_state_heartbeat ON service (state, last_heartbeat); + +PRAGMA optimize;