579 lines
19 KiB
Go
579 lines
19 KiB
Go
// Package store keeps the state of the web UI in SQLite: clusters (root
|
|
// passwords encrypted with AES-256-GCM, pinned TLS certificates and SSH host
|
|
// keys), synchronization settings, per target mappings and policies, users,
|
|
// execution journal and audit trail.
|
|
package store
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"log"
|
|
"strings"
|
|
"time"
|
|
|
|
_ "modernc.org/sqlite" // pure Go driver: the binary stays static (CGO_ENABLED=0)
|
|
|
|
"pve-confsync/internal/adapter"
|
|
"pve-confsync/internal/config"
|
|
"pve-confsync/internal/model"
|
|
"pve-confsync/internal/sshx"
|
|
)
|
|
|
|
// ErrNotFound is returned for an unknown cluster, user or run.
|
|
var ErrNotFound = errors.New("not found")
|
|
|
|
// Store is the SQLite database.
|
|
type Store struct {
|
|
db *sql.DB
|
|
box *Box
|
|
}
|
|
|
|
// schemaVersion 2: clusters reached with the root credentials (API + SSH),
|
|
// roles primary / target / disabled. Version 1 (API tokens) is not migrated:
|
|
// its clusters table is dropped, the clusters are added again from the UI.
|
|
const schemaVersion = 2
|
|
|
|
const schema = `
|
|
CREATE TABLE IF NOT EXISTS clusters (
|
|
name TEXT PRIMARY KEY,
|
|
role TEXT NOT NULL DEFAULT 'target',
|
|
position INTEGER NOT NULL DEFAULT 0,
|
|
url TEXT NOT NULL,
|
|
api_user TEXT NOT NULL DEFAULT 'root@pam',
|
|
password BLOB,
|
|
tls TEXT NOT NULL DEFAULT 'fp',
|
|
fingerprint TEXT NOT NULL DEFAULT '',
|
|
ca_pem TEXT NOT NULL DEFAULT '',
|
|
timeout TEXT NOT NULL DEFAULT '',
|
|
ssh_host TEXT NOT NULL DEFAULT '',
|
|
ssh_port INTEGER NOT NULL DEFAULT 22,
|
|
ssh_hostkey TEXT NOT NULL DEFAULT '',
|
|
mapping TEXT NOT NULL DEFAULT '{}',
|
|
policies TEXT NOT NULL DEFAULT '{}',
|
|
last_check TEXT NOT NULL DEFAULT '',
|
|
updated_at INTEGER NOT NULL
|
|
);
|
|
CREATE TABLE IF NOT EXISTS settings (
|
|
key TEXT PRIMARY KEY,
|
|
value TEXT NOT NULL
|
|
);
|
|
CREATE TABLE IF NOT EXISTS users (
|
|
name TEXT PRIMARY KEY,
|
|
role TEXT NOT NULL,
|
|
password TEXT NOT NULL,
|
|
epoch INTEGER NOT NULL,
|
|
created_at INTEGER NOT NULL,
|
|
last_login INTEGER NOT NULL DEFAULT 0
|
|
);
|
|
CREATE TABLE IF NOT EXISTS runs (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
started_at INTEGER NOT NULL,
|
|
finished_at INTEGER NOT NULL DEFAULT 0,
|
|
user TEXT NOT NULL,
|
|
mode TEXT NOT NULL,
|
|
scope TEXT NOT NULL,
|
|
clients TEXT NOT NULL DEFAULT '',
|
|
objects TEXT NOT NULL DEFAULT '',
|
|
status TEXT NOT NULL,
|
|
summary TEXT NOT NULL DEFAULT '',
|
|
detail TEXT NOT NULL DEFAULT '{}'
|
|
);
|
|
CREATE INDEX IF NOT EXISTS runs_started ON runs(started_at);
|
|
CREATE TABLE IF NOT EXISTS audit (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
at INTEGER NOT NULL,
|
|
user TEXT NOT NULL,
|
|
role TEXT NOT NULL DEFAULT '',
|
|
ip TEXT NOT NULL DEFAULT '',
|
|
action TEXT NOT NULL,
|
|
target TEXT NOT NULL DEFAULT '',
|
|
ok INTEGER NOT NULL DEFAULT 1,
|
|
detail TEXT NOT NULL DEFAULT ''
|
|
);
|
|
CREATE INDEX IF NOT EXISTS audit_at ON audit(at);
|
|
`
|
|
|
|
// Open opens (and creates) the database. key encrypts the token secrets.
|
|
func Open(path string, key []byte) (*Store, error) {
|
|
box, err := NewBox(key)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
// WAL: the UI reads while a synchronization writes the journal.
|
|
db, err := sql.Open("sqlite", "file:"+path+"?_pragma=journal_mode(WAL)&_pragma=busy_timeout(5000)&_pragma=foreign_keys(1)")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
db.SetMaxOpenConns(1) // one writer; SQLite serializes anyway
|
|
if err := migrate(db); err != nil {
|
|
db.Close()
|
|
return nil, fmt.Errorf("%s: %w", path, err)
|
|
}
|
|
if _, err := db.Exec(schema); err != nil {
|
|
db.Close()
|
|
return nil, fmt.Errorf("%s: %w", path, err)
|
|
}
|
|
s := &Store{db: db, box: box}
|
|
if err := s.checkKey(); err != nil {
|
|
db.Close()
|
|
return nil, err
|
|
}
|
|
return s, nil
|
|
}
|
|
|
|
// migrate drops the clusters of the first schema (API tokens, not reusable).
|
|
func migrate(db *sql.DB) error {
|
|
var v int
|
|
if err := db.QueryRow(`PRAGMA user_version`).Scan(&v); err != nil {
|
|
return err
|
|
}
|
|
if v >= schemaVersion {
|
|
return nil
|
|
}
|
|
var old int
|
|
if err := db.QueryRow(`SELECT count(*) FROM pragma_table_info('clusters') WHERE name = 'token_id'`).Scan(&old); err != nil {
|
|
return err
|
|
}
|
|
if old > 0 {
|
|
log.Printf("database: clusters of the previous version (API tokens) dropped, add them again with their root credentials")
|
|
if _, err := db.Exec(`DROP TABLE clusters; DROP TABLE IF EXISTS settings`); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
_, err := db.Exec(fmt.Sprintf(`PRAGMA user_version = %d`, schemaVersion))
|
|
return err
|
|
}
|
|
|
|
// Close closes the database.
|
|
func (s *Store) Close() error { return s.db.Close() }
|
|
|
|
// checkKey refuses to start with a key that does not decrypt the stored secrets
|
|
// (lost or changed CONFSYNC_SECRET_KEY), instead of failing later on each cluster.
|
|
func (s *Store) checkKey() error {
|
|
var blob []byte
|
|
err := s.db.QueryRow(`SELECT password FROM clusters WHERE password IS NOT NULL AND length(password) > 0 LIMIT 1`).Scan(&blob)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return nil
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if _, err := s.box.Open(blob); err != nil {
|
|
return errors.New("the encryption key does not decrypt the stored passwords (CONFSYNC_SECRET_KEY changed?)")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// ---- clusters ----
|
|
|
|
// Cluster roles: one primary (the source, never written), targets (replicated)
|
|
// and disabled clusters (kept, not read).
|
|
const (
|
|
ClusterPrimary = "primary"
|
|
ClusterTarget = "target"
|
|
ClusterDisabled = "disabled"
|
|
)
|
|
|
|
// TLS modes of a cluster.
|
|
const (
|
|
TLSPin = "fp" // pinned certificate (recorded on the first connection when empty)
|
|
TLSCA = "ca" // CA bundle
|
|
TLSSystem = "sys" // system authorities
|
|
TLSInsecure = "insecure" // no verification (lab only)
|
|
)
|
|
|
|
// Check is the result of the last connection test of a cluster.
|
|
type Check struct {
|
|
At time.Time `json:"at"`
|
|
OK bool `json:"ok"`
|
|
Error string `json:"error,omitempty"`
|
|
Warnings []string `json:"warnings,omitempty"`
|
|
Version string `json:"version,omitempty"`
|
|
Nodes []string `json:"nodes,omitempty"`
|
|
API bool `json:"api"`
|
|
SSH bool `json:"ssh"`
|
|
}
|
|
|
|
// Cluster is a cluster as stored. The password is never returned to the UI:
|
|
// HasPassword only says whether one is set.
|
|
type Cluster struct {
|
|
Name string `json:"name"`
|
|
Role string `json:"role"`
|
|
Position int `json:"position"`
|
|
URL string `json:"url"`
|
|
User string `json:"user"`
|
|
Password string `json:"password,omitempty"` // write only
|
|
HasPassword bool `json:"hasPassword"`
|
|
TLS string `json:"tls"`
|
|
Fingerprint string `json:"fingerprint"`
|
|
CAPEM string `json:"caPem"`
|
|
Timeout string `json:"timeout"`
|
|
SSHHost string `json:"sshHost"`
|
|
SSHPort int `json:"sshPort"`
|
|
SSHHostKey string `json:"sshHostKey"` // read only (pinned by the connection test)
|
|
SSHFingerprint string `json:"sshFingerprint"` // read only
|
|
Mapping map[model.Kind]map[string]map[string]any `json:"mapping"`
|
|
Policies map[model.Kind]map[string]adapter.Policy `json:"policies"`
|
|
LastCheck *Check `json:"lastCheck,omitempty"`
|
|
UpdatedAt time.Time `json:"updatedAt"`
|
|
}
|
|
|
|
const clusterCols = `name, role, position, url, api_user, password, tls, fingerprint, ca_pem, timeout,
|
|
ssh_host, ssh_port, ssh_hostkey, mapping, policies, last_check, updated_at`
|
|
|
|
func scanCluster(scan func(...any) error) (Cluster, []byte, error) {
|
|
var c Cluster
|
|
var blob []byte
|
|
var mapping, policies, check string
|
|
var updated int64
|
|
err := scan(&c.Name, &c.Role, &c.Position, &c.URL, &c.User, &blob, &c.TLS, &c.Fingerprint, &c.CAPEM, &c.Timeout,
|
|
&c.SSHHost, &c.SSHPort, &c.SSHHostKey, &mapping, &policies, &check, &updated)
|
|
if err != nil {
|
|
return c, nil, err
|
|
}
|
|
c.HasPassword = len(blob) > 0
|
|
c.SSHFingerprint = sshx.Fingerprint(c.SSHHostKey)
|
|
c.UpdatedAt = time.Unix(updated, 0).UTC()
|
|
_ = json.Unmarshal([]byte(mapping), &c.Mapping)
|
|
_ = json.Unmarshal([]byte(policies), &c.Policies)
|
|
if check != "" {
|
|
c.LastCheck = &Check{}
|
|
_ = json.Unmarshal([]byte(check), c.LastCheck)
|
|
}
|
|
return c, blob, nil
|
|
}
|
|
|
|
// Clusters returns every cluster, primary first, without passwords.
|
|
func (s *Store) Clusters(ctx context.Context) ([]Cluster, error) {
|
|
rows, err := s.db.QueryContext(ctx, `SELECT `+clusterCols+` FROM clusters
|
|
ORDER BY role = 'primary' DESC, position, name`)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
var out []Cluster
|
|
for rows.Next() {
|
|
c, _, err := scanCluster(rows.Scan)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
out = append(out, c)
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
|
|
// Cluster returns one cluster with its decrypted connection (for a test).
|
|
func (s *Store) Cluster(ctx context.Context, name string) (Cluster, config.Cluster, error) {
|
|
c, blob, err := scanCluster(s.db.QueryRowContext(ctx, `SELECT `+clusterCols+` FROM clusters WHERE name = ?`, name).Scan)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return c, config.Cluster{}, ErrNotFound
|
|
}
|
|
if err != nil {
|
|
return c, config.Cluster{}, err
|
|
}
|
|
cc, err := s.connection(c, blob)
|
|
return c, cc, err
|
|
}
|
|
|
|
func (s *Store) connection(c Cluster, blob []byte) (config.Cluster, error) {
|
|
pass, err := s.box.Open(blob)
|
|
if err != nil {
|
|
return config.Cluster{}, fmt.Errorf("cluster %s: password: %w", c.Name, err)
|
|
}
|
|
cc := config.Cluster{Name: c.Name, URL: config.NormalizeURL(c.URL), User: c.User, Password: string(pass), Timeout: c.Timeout,
|
|
SSHHost: c.SSHHost, SSHPort: c.SSHPort, SSHHostKey: c.SSHHostKey}
|
|
switch c.TLS {
|
|
case TLSPin:
|
|
cc.Fingerprint = c.Fingerprint
|
|
case TLSCA:
|
|
cc.CAPEM = c.CAPEM
|
|
case TLSInsecure:
|
|
cc.Insecure = true
|
|
}
|
|
return cc, nil
|
|
}
|
|
|
|
// ValidClusterRole tells whether r is a cluster role.
|
|
func ValidClusterRole(r string) bool {
|
|
return r == ClusterPrimary || r == ClusterTarget || r == ClusterDisabled
|
|
}
|
|
|
|
// SaveCluster creates or updates a cluster. An empty Password keeps the stored
|
|
// one. oldName renames a cluster (empty: same name). A primary cluster makes
|
|
// the previous primary a target. The pinned TLS certificate and SSH host key
|
|
// are forgotten when the address they belong to changes.
|
|
func (s *Store) SaveCluster(ctx context.Context, oldName string, c Cluster) error {
|
|
c.Name = strings.TrimSpace(c.Name)
|
|
c.URL = config.NormalizeURL(c.URL)
|
|
if c.Name == "" || c.URL == "" {
|
|
return errors.New("name and url are required")
|
|
}
|
|
if c.User == "" {
|
|
c.User = "root@pam"
|
|
}
|
|
if c.Role == "" {
|
|
c.Role = ClusterTarget
|
|
}
|
|
if !ValidClusterRole(c.Role) {
|
|
return fmt.Errorf("role %q: primary, target or disabled expected", c.Role)
|
|
}
|
|
switch c.TLS {
|
|
case "":
|
|
c.TLS = TLSPin
|
|
case TLSPin, TLSCA, TLSSystem, TLSInsecure:
|
|
default:
|
|
return fmt.Errorf("tls %q: fp, ca, sys or insecure expected", c.TLS)
|
|
}
|
|
if c.TLS == TLSCA && strings.TrimSpace(c.CAPEM) == "" {
|
|
return errors.New("the CA certificate (PEM) is required")
|
|
}
|
|
if c.SSHPort == 0 {
|
|
c.SSHPort = 22
|
|
}
|
|
if oldName == "" {
|
|
oldName = c.Name
|
|
}
|
|
mapping, _ := json.Marshal(nonNil(c.Mapping))
|
|
policies, _ := json.Marshal(nonNilP(c.Policies))
|
|
tx, err := s.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer tx.Rollback()
|
|
old, blob, err := scanCluster(tx.QueryRowContext(ctx, `SELECT `+clusterCols+` FROM clusters WHERE name = ?`, oldName).Scan)
|
|
exists := err == nil
|
|
if err != nil && !errors.Is(err, sql.ErrNoRows) {
|
|
return err
|
|
}
|
|
if c.Password != "" {
|
|
if blob, err = s.box.Seal([]byte(c.Password)); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
if len(blob) == 0 {
|
|
return errors.New("the root password is required")
|
|
}
|
|
hostKey := ""
|
|
if exists {
|
|
hostKey = old.SSHHostKey
|
|
if old.URL != c.URL && c.Fingerprint == old.Fingerprint {
|
|
c.Fingerprint = "" // another server: pinned again on the next test
|
|
}
|
|
if sshTarget(old) != sshTarget(c) {
|
|
hostKey = ""
|
|
}
|
|
}
|
|
if c.Role == ClusterPrimary {
|
|
if _, err := tx.ExecContext(ctx, `UPDATE clusters SET role = 'target' WHERE role = 'primary'`); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
now := time.Now().Unix()
|
|
if exists {
|
|
_, err = tx.ExecContext(ctx, `UPDATE clusters SET name=?, role=?, position=?, url=?, api_user=?, password=?, tls=?,
|
|
fingerprint=?, ca_pem=?, timeout=?, ssh_host=?, ssh_port=?, ssh_hostkey=?, mapping=?, policies=?, updated_at=?
|
|
WHERE name=?`,
|
|
c.Name, c.Role, c.Position, c.URL, c.User, blob, c.TLS, c.Fingerprint, c.CAPEM, c.Timeout,
|
|
c.SSHHost, c.SSHPort, hostKey, string(mapping), string(policies), now, oldName)
|
|
} else {
|
|
_, err = tx.ExecContext(ctx, `INSERT INTO clusters (name, role, position, url, api_user, password, tls, fingerprint,
|
|
ca_pem, timeout, ssh_host, ssh_port, ssh_hostkey, mapping, policies, updated_at)
|
|
VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`,
|
|
c.Name, c.Role, c.Position, c.URL, c.User, blob, c.TLS, c.Fingerprint, c.CAPEM, c.Timeout,
|
|
c.SSHHost, c.SSHPort, "", string(mapping), string(policies), now)
|
|
}
|
|
if err != nil {
|
|
if strings.Contains(err.Error(), "UNIQUE") {
|
|
return fmt.Errorf("cluster %q already exists", c.Name)
|
|
}
|
|
return err
|
|
}
|
|
return tx.Commit()
|
|
}
|
|
|
|
// sshTarget is the SSH endpoint of a cluster, to detect a change of node.
|
|
func sshTarget(c Cluster) string {
|
|
host := c.SSHHost
|
|
if host == "" {
|
|
host = c.URL
|
|
}
|
|
return fmt.Sprintf("%s|%d", host, c.SSHPort)
|
|
}
|
|
|
|
// SetRole changes the role of a cluster (a new primary demotes the previous one).
|
|
func (s *Store) SetRole(ctx context.Context, name, role string) error {
|
|
if !ValidClusterRole(role) {
|
|
return fmt.Errorf("role %q: primary, target or disabled expected", role)
|
|
}
|
|
tx, err := s.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer tx.Rollback()
|
|
if role == ClusterPrimary {
|
|
if _, err := tx.ExecContext(ctx, `UPDATE clusters SET role = 'target' WHERE role = 'primary' AND name <> ?`, name); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
res, err := tx.ExecContext(ctx, `UPDATE clusters SET role = ?, updated_at = ? WHERE name = ?`, role, time.Now().Unix(), name)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if n, _ := res.RowsAffected(); n == 0 {
|
|
return ErrNotFound
|
|
}
|
|
return tx.Commit()
|
|
}
|
|
|
|
// Pin records the TLS fingerprint and/or the SSH host key seen on the first
|
|
// connection (empty values are left unchanged), and the result of the test.
|
|
func (s *Store) Pin(ctx context.Context, name, fingerprint, hostKey string, check *Check) error {
|
|
var chk []byte
|
|
if check != nil {
|
|
chk, _ = json.Marshal(check)
|
|
}
|
|
_, err := s.db.ExecContext(ctx, `UPDATE clusters SET
|
|
fingerprint = CASE WHEN ? <> '' THEN ? ELSE fingerprint END,
|
|
ssh_hostkey = CASE WHEN ? <> '' THEN ? ELSE ssh_hostkey END,
|
|
last_check = CASE WHEN ? <> '' THEN ? ELSE last_check END
|
|
WHERE name = ?`, fingerprint, fingerprint, hostKey, hostKey, string(chk), string(chk), name)
|
|
return err
|
|
}
|
|
|
|
// Forget clears the pinned TLS fingerprint and SSH host key (pinned again on the next test).
|
|
func (s *Store) Forget(ctx context.Context, name string) error {
|
|
res, err := s.db.ExecContext(ctx, `UPDATE clusters SET fingerprint = CASE WHEN tls = 'fp' THEN '' ELSE fingerprint END,
|
|
ssh_hostkey = '' WHERE name = ?`, name)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if n, _ := res.RowsAffected(); n == 0 {
|
|
return ErrNotFound
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// DeleteCluster removes a cluster.
|
|
func (s *Store) DeleteCluster(ctx context.Context, name string) error {
|
|
res, err := s.db.ExecContext(ctx, `DELETE FROM clusters WHERE name = ?`, name)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if n, _ := res.RowsAffected(); n == 0 {
|
|
return ErrNotFound
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func nonNil(m map[model.Kind]map[string]map[string]any) any {
|
|
if m == nil {
|
|
return map[string]any{}
|
|
}
|
|
return m
|
|
}
|
|
|
|
func nonNilP(m map[model.Kind]map[string]adapter.Policy) any {
|
|
if m == nil {
|
|
return map[string]any{}
|
|
}
|
|
return m
|
|
}
|
|
|
|
// ---- settings ----
|
|
|
|
// Settings are the global synchronization settings (the rest of the YAML file).
|
|
type Settings struct {
|
|
Kinds []model.Kind `json:"kinds"`
|
|
Objects map[model.Kind]config.Selection `json:"objects"`
|
|
Policies map[model.Kind]map[string]adapter.Policy `json:"policies"`
|
|
Prune bool `json:"prune"`
|
|
}
|
|
|
|
// Settings returns the global settings (all kinds by default).
|
|
func (s *Store) Settings(ctx context.Context) (Settings, error) {
|
|
var st Settings
|
|
var raw string
|
|
err := s.db.QueryRowContext(ctx, `SELECT value FROM settings WHERE key = 'sync'`).Scan(&raw)
|
|
switch {
|
|
case errors.Is(err, sql.ErrNoRows):
|
|
case err != nil:
|
|
return st, err
|
|
default:
|
|
if err := json.Unmarshal([]byte(raw), &st); err != nil {
|
|
return st, err
|
|
}
|
|
}
|
|
if len(st.Kinds) == 0 {
|
|
st.Kinds = append(st.Kinds, model.Kinds...)
|
|
}
|
|
// empty maps, not null, for the web UI
|
|
if st.Objects == nil {
|
|
st.Objects = map[model.Kind]config.Selection{}
|
|
}
|
|
if st.Policies == nil {
|
|
st.Policies = map[model.Kind]map[string]adapter.Policy{}
|
|
}
|
|
return st, nil
|
|
}
|
|
|
|
// SaveSettings replaces the global settings.
|
|
func (s *Store) SaveSettings(ctx context.Context, st Settings) error {
|
|
b, err := json.Marshal(st)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
_, err = s.db.ExecContext(ctx, `INSERT INTO settings (key, value) VALUES ('sync', ?)
|
|
ON CONFLICT(key) DO UPDATE SET value = excluded.value`, string(b))
|
|
return err
|
|
}
|
|
|
|
// ---- configuration for the engine ----
|
|
|
|
// Config builds the engine configuration, passwords decrypted: the primary
|
|
// cluster and the targets (disabled clusters are left aside). It fails when no
|
|
// primary is defined.
|
|
func (s *Store) Config(ctx context.Context) (*config.Config, error) {
|
|
st, err := s.Settings(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
cfg := &config.Config{Kinds: st.Kinds, Objects: st.Objects, Policies: st.Policies, Prune: st.Prune}
|
|
rows, err := s.db.QueryContext(ctx, `SELECT `+clusterCols+` FROM clusters ORDER BY position, name`)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
hasRef := false
|
|
for rows.Next() {
|
|
c, blob, err := scanCluster(rows.Scan)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if c.Role == ClusterDisabled {
|
|
continue
|
|
}
|
|
cc, err := s.connection(c, blob)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if c.Role == ClusterPrimary {
|
|
cfg.Reference, hasRef = cc, true
|
|
continue
|
|
}
|
|
cfg.Clients = append(cfg.Clients, config.Client{Cluster: cc, Mapping: c.Mapping, Policies: c.Policies})
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
if !hasRef {
|
|
return nil, ErrNoReference
|
|
}
|
|
return cfg, cfg.Validate()
|
|
}
|
|
|
|
// ErrNoReference: nothing can be compared before a primary cluster is chosen.
|
|
var ErrNoReference = errors.New("no primary cluster defined")
|