Implement SQLite database persistence for instance management

This commit is contained in:
2025-11-30 00:12:03 +01:00
parent 0c11365d7e
commit fec989fee2
16 changed files with 1325 additions and 245 deletions

View File

@@ -9,6 +9,7 @@ import (
"runtime"
"strconv"
"strings"
"time"
"gopkg.in/yaml.v3"
)
@@ -42,6 +43,7 @@ type AppConfig struct {
Server ServerConfig `yaml:"server" json:"server"`
Backends BackendConfig `yaml:"backends" json:"backends"`
Instances InstancesConfig `yaml:"instances" json:"instances"`
Database DatabaseConfig `yaml:"database" json:"database"`
Auth AuthConfig `yaml:"auth" json:"auth"`
LocalNode string `yaml:"local_node,omitempty" json:"local_node,omitempty"`
Nodes map[string]NodeConfig `yaml:"nodes,omitempty" json:"nodes,omitempty"`
@@ -71,6 +73,17 @@ type ServerConfig struct {
ResponseHeaders map[string]string `yaml:"response_headers,omitempty" json:"response_headers,omitempty"`
}
// DatabaseConfig contains database configuration settings
type DatabaseConfig struct {
// Database file path (relative to data_dir or absolute)
Path string `yaml:"path" json:"path"`
// Connection settings
MaxOpenConnections int `yaml:"max_open_connections" json:"max_open_connections"`
MaxIdleConnections int `yaml:"max_idle_connections" json:"max_idle_connections"`
ConnMaxLifetime time.Duration `yaml:"connection_max_lifetime" json:"connection_max_lifetime"`
}
// InstancesConfig contains instance management configuration
type InstancesConfig struct {
// Port range for instances (e.g., 8000,9000)
@@ -204,6 +217,12 @@ func LoadConfig(configPath string) (AppConfig, error) {
OnDemandStartTimeout: 120, // 2 minutes
TimeoutCheckInterval: 5, // Check timeouts every 5 minutes
},
Database: DatabaseConfig{
Path: "llamactl.db", // Relative to data_dir
MaxOpenConnections: 25,
MaxIdleConnections: 5,
ConnMaxLifetime: 5 * time.Minute,
},
Auth: AuthConfig{
RequireInferenceAuth: true,
InferenceKeys: []string{},
@@ -233,6 +252,11 @@ func LoadConfig(configPath string) (AppConfig, error) {
cfg.Instances.LogsDir = filepath.Join(cfg.Instances.DataDir, "logs")
}
// Resolve database path relative to DataDir if it's not absolute
if cfg.Database.Path != "" && !filepath.IsAbs(cfg.Database.Path) {
cfg.Database.Path = filepath.Join(cfg.Instances.DataDir, cfg.Database.Path)
}
// Validate port range
if cfg.Instances.PortRange[0] <= 0 || cfg.Instances.PortRange[1] <= 0 || cfg.Instances.PortRange[0] >= cfg.Instances.PortRange[1] {
return AppConfig{}, fmt.Errorf("invalid port range: %v", cfg.Instances.PortRange)
@@ -495,6 +519,26 @@ func loadEnvVars(cfg *AppConfig) {
if localNode := os.Getenv("LLAMACTL_LOCAL_NODE"); localNode != "" {
cfg.LocalNode = localNode
}
// Database config
if dbPath := os.Getenv("LLAMACTL_DATABASE_PATH"); dbPath != "" {
cfg.Database.Path = dbPath
}
if maxOpenConns := os.Getenv("LLAMACTL_DATABASE_MAX_OPEN_CONNECTIONS"); maxOpenConns != "" {
if m, err := strconv.Atoi(maxOpenConns); err == nil {
cfg.Database.MaxOpenConnections = m
}
}
if maxIdleConns := os.Getenv("LLAMACTL_DATABASE_MAX_IDLE_CONNECTIONS"); maxIdleConns != "" {
if m, err := strconv.Atoi(maxIdleConns); err == nil {
cfg.Database.MaxIdleConnections = m
}
}
if connMaxLifetime := os.Getenv("LLAMACTL_DATABASE_CONN_MAX_LIFETIME"); connMaxLifetime != "" {
if d, err := time.ParseDuration(connMaxLifetime); err == nil {
cfg.Database.ConnMaxLifetime = d
}
}
}
// ParsePortRange parses port range from string formats like "8000-9000" or "8000,9000"

107
pkg/database/database.go Normal file
View File

@@ -0,0 +1,107 @@
package database
import (
"database/sql"
"fmt"
"llamactl/pkg/instance"
"log"
"path/filepath"
"time"
_ "github.com/mattn/go-sqlite3"
)
// Database defines the interface for instance persistence operations
type Database interface {
Save(inst *instance.Instance) error
Delete(name string) error
LoadAll() ([]*instance.Instance, error)
Close() error
}
// Config contains database configuration settings
type Config struct {
// Database file path (relative to data_dir or absolute)
Path string
// Connection settings
MaxOpenConnections int
MaxIdleConnections int
ConnMaxLifetime time.Duration
}
// DB wraps the database connection with configuration
type DB struct {
*sql.DB
config *Config
}
// Open creates a new database connection with the provided configuration
func Open(config *Config) (*DB, error) {
if config == nil {
return nil, fmt.Errorf("database config cannot be nil")
}
if config.Path == "" {
return nil, fmt.Errorf("database path cannot be empty")
}
// Ensure the database directory exists
dbDir := filepath.Dir(config.Path)
if dbDir != "." && dbDir != "/" {
// Directory will be created by the manager if auto_create_dirs is enabled
log.Printf("Database will be created at: %s", config.Path)
}
// Open SQLite database with proper options
// - _journal_mode=WAL: Write-Ahead Logging for better concurrency
// - _busy_timeout=5000: Wait up to 5 seconds if database is locked
// - _foreign_keys=1: Enable foreign key constraints (for future use)
dsn := fmt.Sprintf("file:%s?_journal_mode=WAL&_busy_timeout=5000&_foreign_keys=1", config.Path)
sqlDB, err := sql.Open("sqlite3", dsn)
if err != nil {
return nil, fmt.Errorf("failed to open database: %w", err)
}
// Configure connection pool
if config.MaxOpenConnections > 0 {
sqlDB.SetMaxOpenConns(config.MaxOpenConnections)
}
if config.MaxIdleConnections > 0 {
sqlDB.SetMaxIdleConns(config.MaxIdleConnections)
}
if config.ConnMaxLifetime > 0 {
sqlDB.SetConnMaxLifetime(config.ConnMaxLifetime)
}
// Verify database connection
if err := sqlDB.Ping(); err != nil {
sqlDB.Close()
return nil, fmt.Errorf("failed to ping database: %w", err)
}
log.Printf("Database connection established: %s", config.Path)
return &DB{
DB: sqlDB,
config: config,
}, nil
}
// Close closes the database connection
func (db *DB) Close() error {
if db.DB != nil {
log.Println("Closing database connection")
return db.DB.Close()
}
return nil
}
// HealthCheck verifies the database is accessible
func (db *DB) HealthCheck() error {
if db.DB == nil {
return fmt.Errorf("database connection is nil")
}
return db.DB.Ping()
}

539
pkg/database/instances.go Normal file
View File

@@ -0,0 +1,539 @@
package database
import (
"context"
"database/sql"
"encoding/json"
"fmt"
"llamactl/pkg/backends"
"llamactl/pkg/instance"
"log"
"time"
)
// instanceRow represents a row in the instances table
type instanceRow struct {
ID int
Name string
BackendType string
BackendConfigJSON string
Status string
CreatedAt int64
UpdatedAt int64
AutoRestart int
MaxRestarts int
RestartDelay int
OnDemandStart int
IdleTimeout int
DockerEnabled int
CommandOverride sql.NullString
Nodes sql.NullString
Environment sql.NullString
OwnerUserID sql.NullString
}
// Create inserts a new instance into the database
func (db *DB) Create(ctx context.Context, inst *instance.Instance) error {
if inst == nil {
return fmt.Errorf("instance cannot be nil")
}
opts := inst.GetOptions()
if opts == nil {
return fmt.Errorf("instance options cannot be nil")
}
// Convert instance to database row
row, err := db.instanceToRow(inst)
if err != nil {
return fmt.Errorf("failed to convert instance to row: %w", err)
}
// Insert into database
query := `
INSERT INTO instances (
name, backend_type, backend_config_json, status,
created_at, updated_at,
auto_restart, max_restarts, restart_delay,
on_demand_start, idle_timeout, docker_enabled,
command_override, nodes, environment, owner_user_id
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
`
_, err = db.DB.ExecContext(ctx, query,
row.Name, row.BackendType, row.BackendConfigJSON, row.Status,
row.CreatedAt, row.UpdatedAt,
row.AutoRestart, row.MaxRestarts, row.RestartDelay,
row.OnDemandStart, row.IdleTimeout, row.DockerEnabled,
row.CommandOverride, row.Nodes, row.Environment, row.OwnerUserID,
)
if err != nil {
return fmt.Errorf("failed to insert instance: %w", err)
}
return nil
}
// GetByName retrieves an instance by name
func (db *DB) GetByName(ctx context.Context, name string) (*instance.Instance, error) {
query := `
SELECT id, name, backend_type, backend_config_json, status,
created_at, updated_at,
auto_restart, max_restarts, restart_delay,
on_demand_start, idle_timeout, docker_enabled,
command_override, nodes, environment, owner_user_id
FROM instances
WHERE name = ?
`
var row instanceRow
err := db.DB.QueryRowContext(ctx, query, name).Scan(
&row.ID, &row.Name, &row.BackendType, &row.BackendConfigJSON, &row.Status,
&row.CreatedAt, &row.UpdatedAt,
&row.AutoRestart, &row.MaxRestarts, &row.RestartDelay,
&row.OnDemandStart, &row.IdleTimeout, &row.DockerEnabled,
&row.CommandOverride, &row.Nodes, &row.Environment, &row.OwnerUserID,
)
if err == sql.ErrNoRows {
return nil, fmt.Errorf("instance not found: %s", name)
}
if err != nil {
return nil, fmt.Errorf("failed to query instance: %w", err)
}
return db.rowToInstance(&row)
}
// GetAll retrieves all instances from the database
func (db *DB) GetAll(ctx context.Context) ([]*instance.Instance, error) {
query := `
SELECT id, name, backend_type, backend_config_json, status,
created_at, updated_at,
auto_restart, max_restarts, restart_delay,
on_demand_start, idle_timeout, docker_enabled,
command_override, nodes, environment, owner_user_id
FROM instances
ORDER BY created_at ASC
`
rows, err := db.DB.QueryContext(ctx, query)
if err != nil {
return nil, fmt.Errorf("failed to query instances: %w", err)
}
defer rows.Close()
var instances []*instance.Instance
for rows.Next() {
var row instanceRow
err := rows.Scan(
&row.ID, &row.Name, &row.BackendType, &row.BackendConfigJSON, &row.Status,
&row.CreatedAt, &row.UpdatedAt,
&row.AutoRestart, &row.MaxRestarts, &row.RestartDelay,
&row.OnDemandStart, &row.IdleTimeout, &row.DockerEnabled,
&row.CommandOverride, &row.Nodes, &row.Environment, &row.OwnerUserID,
)
if err != nil {
log.Printf("Failed to scan instance row: %v", err)
continue
}
inst, err := db.rowToInstance(&row)
if err != nil {
log.Printf("Failed to convert row to instance: %v", err)
continue
}
instances = append(instances, inst)
}
if err := rows.Err(); err != nil {
return nil, fmt.Errorf("error iterating rows: %w", err)
}
return instances, nil
}
// Update updates an existing instance
func (db *DB) Update(ctx context.Context, inst *instance.Instance) error {
if inst == nil {
return fmt.Errorf("instance cannot be nil")
}
opts := inst.GetOptions()
if opts == nil {
return fmt.Errorf("instance options cannot be nil")
}
// Convert instance to database row
row, err := db.instanceToRow(inst)
if err != nil {
return fmt.Errorf("failed to convert instance to row: %w", err)
}
// Update in database
query := `
UPDATE instances SET
backend_type = ?, backend_config_json = ?, status = ?,
updated_at = ?,
auto_restart = ?, max_restarts = ?, restart_delay = ?,
on_demand_start = ?, idle_timeout = ?, docker_enabled = ?,
command_override = ?, nodes = ?, environment = ?
WHERE name = ?
`
result, err := db.DB.ExecContext(ctx, query,
row.BackendType, row.BackendConfigJSON, row.Status,
row.UpdatedAt,
row.AutoRestart, row.MaxRestarts, row.RestartDelay,
row.OnDemandStart, row.IdleTimeout, row.DockerEnabled,
row.CommandOverride, row.Nodes, row.Environment,
row.Name,
)
if err != nil {
return fmt.Errorf("failed to update instance: %w", err)
}
rowsAffected, err := result.RowsAffected()
if err != nil {
return fmt.Errorf("failed to get rows affected: %w", err)
}
if rowsAffected == 0 {
return fmt.Errorf("instance not found: %s", inst.Name)
}
return nil
}
// UpdateStatus updates only the status of an instance (optimized operation)
func (db *DB) UpdateStatus(ctx context.Context, name string, status instance.Status) error {
// Convert status to string
statusJSON, err := status.MarshalJSON()
if err != nil {
return fmt.Errorf("failed to marshal status: %w", err)
}
var statusStr string
if err := json.Unmarshal(statusJSON, &statusStr); err != nil {
return fmt.Errorf("failed to unmarshal status string: %w", err)
}
query := `
UPDATE instances SET
status = ?,
updated_at = ?
WHERE name = ?
`
result, err := db.DB.ExecContext(ctx, query, statusStr, time.Now().Unix(), name)
if err != nil {
return fmt.Errorf("failed to update instance status: %w", err)
}
rowsAffected, err := result.RowsAffected()
if err != nil {
return fmt.Errorf("failed to get rows affected: %w", err)
}
if rowsAffected == 0 {
return fmt.Errorf("instance not found: %s", name)
}
return nil
}
// DeleteInstance removes an instance from the database
func (db *DB) DeleteInstance(ctx context.Context, name string) error {
query := `DELETE FROM instances WHERE name = ?`
result, err := db.DB.ExecContext(ctx, query, name)
if err != nil {
return fmt.Errorf("failed to delete instance: %w", err)
}
rowsAffected, err := result.RowsAffected()
if err != nil {
return fmt.Errorf("failed to get rows affected: %w", err)
}
if rowsAffected == 0 {
return fmt.Errorf("instance not found: %s", name)
}
return nil
}
// instanceToRow converts an Instance to a database row
func (db *DB) instanceToRow(inst *instance.Instance) (*instanceRow, error) {
opts := inst.GetOptions()
if opts == nil {
return nil, fmt.Errorf("instance options cannot be nil")
}
// Marshal backend options to JSON (this uses the MarshalJSON method which handles typed backends)
backendJSON, err := json.Marshal(&opts.BackendOptions)
if err != nil {
return nil, fmt.Errorf("failed to marshal backend options: %w", err)
}
// Extract just the backend_options field from the marshaled JSON
var backendWrapper struct {
BackendOptions map[string]any `json:"backend_options"`
}
if err := json.Unmarshal(backendJSON, &backendWrapper); err != nil {
return nil, fmt.Errorf("failed to unmarshal backend wrapper: %w", err)
}
backendConfigJSON, err := json.Marshal(backendWrapper.BackendOptions)
if err != nil {
return nil, fmt.Errorf("failed to marshal backend config: %w", err)
}
// Convert nodes map to JSON array
var nodesJSON sql.NullString
if len(opts.Nodes) > 0 {
nodesList := make([]string, 0, len(opts.Nodes))
for node := range opts.Nodes {
nodesList = append(nodesList, node)
}
nodesBytes, err := json.Marshal(nodesList)
if err != nil {
return nil, fmt.Errorf("failed to marshal nodes: %w", err)
}
nodesJSON = sql.NullString{String: string(nodesBytes), Valid: true}
}
// Convert environment map to JSON
var envJSON sql.NullString
if len(opts.Environment) > 0 {
envBytes, err := json.Marshal(opts.Environment)
if err != nil {
return nil, fmt.Errorf("failed to marshal environment: %w", err)
}
envJSON = sql.NullString{String: string(envBytes), Valid: true}
}
// Convert command override
var cmdOverride sql.NullString
if opts.CommandOverride != "" {
cmdOverride = sql.NullString{String: opts.CommandOverride, Valid: true}
}
// Convert boolean pointers to integers (0 or 1)
autoRestart := 0
if opts.AutoRestart != nil && *opts.AutoRestart {
autoRestart = 1
}
maxRestarts := -1
if opts.MaxRestarts != nil {
maxRestarts = *opts.MaxRestarts
}
restartDelay := 0
if opts.RestartDelay != nil {
restartDelay = *opts.RestartDelay
}
onDemandStart := 0
if opts.OnDemandStart != nil && *opts.OnDemandStart {
onDemandStart = 1
}
idleTimeout := 0
if opts.IdleTimeout != nil {
idleTimeout = *opts.IdleTimeout
}
dockerEnabled := 0
if opts.DockerEnabled != nil && *opts.DockerEnabled {
dockerEnabled = 1
}
now := time.Now().Unix()
// Convert status to string
statusJSON, err := inst.GetStatus().MarshalJSON()
if err != nil {
return nil, fmt.Errorf("failed to marshal status: %w", err)
}
var statusStr string
if err := json.Unmarshal(statusJSON, &statusStr); err != nil {
return nil, fmt.Errorf("failed to unmarshal status string: %w", err)
}
return &instanceRow{
Name: inst.Name,
BackendType: string(opts.BackendOptions.BackendType),
BackendConfigJSON: string(backendConfigJSON),
Status: statusStr,
CreatedAt: inst.Created,
UpdatedAt: now,
AutoRestart: autoRestart,
MaxRestarts: maxRestarts,
RestartDelay: restartDelay,
OnDemandStart: onDemandStart,
IdleTimeout: idleTimeout,
DockerEnabled: dockerEnabled,
CommandOverride: cmdOverride,
Nodes: nodesJSON,
Environment: envJSON,
}, nil
}
// rowToInstance converts a database row to an Instance
func (db *DB) rowToInstance(row *instanceRow) (*instance.Instance, error) {
// Unmarshal backend config
var backendConfig map[string]any
if err := json.Unmarshal([]byte(row.BackendConfigJSON), &backendConfig); err != nil {
return nil, fmt.Errorf("failed to unmarshal backend config: %w", err)
}
// Create backends.Options by marshaling and unmarshaling to trigger the UnmarshalJSON logic
// This ensures the typed backend fields (LlamaServerOptions, VllmServerOptions, etc.) are populated
var backendOptions backends.Options
backendJSON, err := json.Marshal(map[string]any{
"backend_type": row.BackendType,
"backend_options": backendConfig,
})
if err != nil {
return nil, fmt.Errorf("failed to marshal backend for unmarshaling: %w", err)
}
if err := json.Unmarshal(backendJSON, &backendOptions); err != nil {
return nil, fmt.Errorf("failed to unmarshal backend options: %w", err)
}
// Unmarshal nodes
var nodes map[string]struct{}
if row.Nodes.Valid && row.Nodes.String != "" {
var nodesList []string
if err := json.Unmarshal([]byte(row.Nodes.String), &nodesList); err != nil {
return nil, fmt.Errorf("failed to unmarshal nodes: %w", err)
}
nodes = make(map[string]struct{}, len(nodesList))
for _, node := range nodesList {
nodes[node] = struct{}{}
}
}
// Unmarshal environment
var environment map[string]string
if row.Environment.Valid && row.Environment.String != "" {
if err := json.Unmarshal([]byte(row.Environment.String), &environment); err != nil {
return nil, fmt.Errorf("failed to unmarshal environment: %w", err)
}
}
// Convert integers to boolean pointers
autoRestart := row.AutoRestart == 1
maxRestarts := row.MaxRestarts
restartDelay := row.RestartDelay
onDemandStart := row.OnDemandStart == 1
idleTimeout := row.IdleTimeout
dockerEnabled := row.DockerEnabled == 1
// Create instance options
opts := &instance.Options{
AutoRestart: &autoRestart,
MaxRestarts: &maxRestarts,
RestartDelay: &restartDelay,
OnDemandStart: &onDemandStart,
IdleTimeout: &idleTimeout,
DockerEnabled: &dockerEnabled,
CommandOverride: row.CommandOverride.String,
Nodes: nodes,
Environment: environment,
BackendOptions: backendOptions,
}
// Create instance struct and manually unmarshal fields
// We do this manually because BackendOptions and Nodes have json:"-" tags
// and would be lost if we used the marshal/unmarshal cycle
inst := &instance.Instance{
Name: row.Name,
Created: row.CreatedAt,
}
// Create a temporary struct for unmarshaling the status and simple fields
type instanceAux struct {
Name string `json:"name"`
Created int64 `json:"created"`
Status string `json:"status"`
Options struct {
AutoRestart *bool `json:"auto_restart,omitempty"`
MaxRestarts *int `json:"max_restarts,omitempty"`
RestartDelay *int `json:"restart_delay,omitempty"`
OnDemandStart *bool `json:"on_demand_start,omitempty"`
IdleTimeout *int `json:"idle_timeout,omitempty"`
DockerEnabled *bool `json:"docker_enabled,omitempty"`
CommandOverride string `json:"command_override,omitempty"`
Environment map[string]string `json:"environment,omitempty"`
} `json:"options"`
}
aux := instanceAux{
Name: row.Name,
Created: row.CreatedAt,
Status: row.Status,
}
aux.Options.AutoRestart = opts.AutoRestart
aux.Options.MaxRestarts = opts.MaxRestarts
aux.Options.RestartDelay = opts.RestartDelay
aux.Options.OnDemandStart = opts.OnDemandStart
aux.Options.IdleTimeout = opts.IdleTimeout
aux.Options.DockerEnabled = opts.DockerEnabled
aux.Options.CommandOverride = opts.CommandOverride
aux.Options.Environment = opts.Environment
instJSON, err := json.Marshal(aux)
if err != nil {
return nil, fmt.Errorf("failed to marshal instance: %w", err)
}
if err := json.Unmarshal(instJSON, inst); err != nil {
return nil, fmt.Errorf("failed to unmarshal instance: %w", err)
}
// Manually set the fields that have json:"-" tags by using SetOptions
// We need to set the whole options object because GetOptions returns a copy
// and we need to ensure BackendOptions and Nodes (which have json:"-") are set
inst.SetOptions(opts)
return inst, nil
}
// Database interface implementation
// Save saves an instance to the database (insert or update)
func (db *DB) Save(inst *instance.Instance) error {
ctx := context.Background()
// Try to get existing instance
existing, err := db.GetByName(ctx, inst.Name)
if err != nil {
// Instance doesn't exist, create it
return db.Create(ctx, inst)
}
// Instance exists, update it
if existing != nil {
return db.Update(ctx, inst)
}
return db.Create(ctx, inst)
}
// Delete removes an instance from the database
func (db *DB) Delete(name string) error {
ctx := context.Background()
return db.DeleteInstance(ctx, name)
}
// LoadAll loads all instances from the database
func (db *DB) LoadAll() ([]*instance.Instance, error) {
ctx := context.Background()
return db.GetAll(ctx)
}

View File

@@ -0,0 +1,78 @@
package database
import (
"embed"
"fmt"
"log"
"github.com/golang-migrate/migrate/v4"
"github.com/golang-migrate/migrate/v4/database/sqlite3"
"github.com/golang-migrate/migrate/v4/source/iofs"
)
//go:embed migrations/*.sql
var migrationFiles embed.FS
// RunMigrations applies all pending database migrations
func RunMigrations(db *DB) error {
if db == nil || db.DB == nil {
return fmt.Errorf("database connection is nil")
}
// Create migration source from embedded files
sourceDriver, err := iofs.New(migrationFiles, "migrations")
if err != nil {
return fmt.Errorf("failed to create migration source: %w", err)
}
// Create database driver for migrations
dbDriver, err := sqlite3.WithInstance(db.DB, &sqlite3.Config{})
if err != nil {
return fmt.Errorf("failed to create database driver: %w", err)
}
// Create migrator
migrator, err := migrate.NewWithInstance("iofs", sourceDriver, "sqlite3", dbDriver)
if err != nil {
return fmt.Errorf("failed to create migrator: %w", err)
}
// Get current version
currentVersion, dirty, err := migrator.Version()
if err != nil && err != migrate.ErrNilVersion {
return fmt.Errorf("failed to get current migration version: %w", err)
}
if dirty {
return fmt.Errorf("database is in dirty state at version %d - manual intervention required", currentVersion)
}
// Run migrations
log.Printf("Running database migrations (current version: %v)", currentVersionString(currentVersion, err))
if err := migrator.Up(); err != nil {
if err == migrate.ErrNoChange {
log.Println("Database schema is up to date")
return nil
}
return fmt.Errorf("failed to run migrations: %w", err)
}
// Get new version
newVersion, _, err := migrator.Version()
if err != nil {
log.Printf("Migrations completed (unable to determine new version: %v)", err)
} else {
log.Printf("Migrations completed successfully (new version: %d)", newVersion)
}
return nil
}
// currentVersionString returns a string representation of the current version
func currentVersionString(version uint, err error) string {
if err == migrate.ErrNilVersion {
return "none"
}
return fmt.Sprintf("%d", version)
}

View File

@@ -0,0 +1,7 @@
-- Drop indexes first
DROP INDEX IF EXISTS idx_instances_backend_type;
DROP INDEX IF EXISTS idx_instances_status;
DROP INDEX IF EXISTS idx_instances_name;
-- Drop tables
DROP TABLE IF EXISTS instances;

View File

@@ -0,0 +1,43 @@
-- -----------------------------------------------------------------------------
-- Instances Table: Central configuration and state for LLM backends
-- -----------------------------------------------------------------------------
CREATE TABLE IF NOT EXISTS instances (
-- Primary identification
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL UNIQUE,
-- Backend configuration
backend_type TEXT NOT NULL CHECK(backend_type IN ('llama_cpp', 'mlx_lm', 'vllm')),
backend_config_json TEXT NOT NULL, -- Backend-specific options (150+ fields for llama_cpp, etc.)
-- Instance state
status TEXT NOT NULL CHECK(status IN ('stopped', 'running', 'failed', 'restarting', 'shutting_down')) DEFAULT 'stopped',
-- Timestamps (created_at stored as Unix timestamp for compatibility with existing JSON format)
created_at INTEGER NOT NULL,
updated_at INTEGER NOT NULL,
-- Common instance options (extracted from Instance.Options)
-- NOT NULL with defaults to match config behavior (nil pointers use these defaults)
auto_restart INTEGER NOT NULL DEFAULT 0, -- Boolean: Enable automatic restart on failure
max_restarts INTEGER NOT NULL DEFAULT -1, -- Maximum restart attempts (-1 = unlimited)
restart_delay INTEGER NOT NULL DEFAULT 0, -- Delay between restarts in seconds
on_demand_start INTEGER NOT NULL DEFAULT 0, -- Boolean: Enable on-demand instance start
idle_timeout INTEGER NOT NULL DEFAULT 0, -- Idle timeout in minutes before auto-stop
docker_enabled INTEGER NOT NULL DEFAULT 0, -- Boolean: Run instance in Docker container
command_override TEXT, -- Custom command to override default backend command (nullable)
-- JSON fields for complex structures (nullable - empty when not set)
nodes TEXT, -- JSON array of node names for remote instances
environment TEXT, -- JSON map of environment variables
-- Future extensibility hook
owner_user_id TEXT NULL -- Future: OIDC user ID for ownership
);
-- -----------------------------------------------------------------------------
-- Indexes for performance
-- -----------------------------------------------------------------------------
CREATE UNIQUE INDEX IF NOT EXISTS idx_instances_name ON instances(name);
CREATE INDEX IF NOT EXISTS idx_instances_status ON instances(status);
CREATE INDEX IF NOT EXISTS idx_instances_backend_type ON instances(backend_type);

View File

@@ -4,6 +4,7 @@ import (
"context"
"fmt"
"llamactl/pkg/config"
"llamactl/pkg/database"
"llamactl/pkg/instance"
"log"
"sync"
@@ -28,11 +29,11 @@ type InstanceManager interface {
type instanceManager struct {
// Components (each with own synchronization)
registry *instanceRegistry
ports *portAllocator
persistence *instancePersister
remote *remoteManager
lifecycle *lifecycleManager
registry *instanceRegistry
ports *portAllocator
db database.Database
remote *remoteManager
lifecycle *lifecycleManager
// Configuration
globalConfig *config.AppConfig
@@ -42,8 +43,8 @@ type instanceManager struct {
shutdownOnce sync.Once
}
// New creates a new instance of InstanceManager.
func New(globalConfig *config.AppConfig) InstanceManager {
// New creates a new instance of InstanceManager with dependency injection.
func New(globalConfig *config.AppConfig, db database.Database) InstanceManager {
if globalConfig.Instances.TimeoutCheckInterval <= 0 {
globalConfig.Instances.TimeoutCheckInterval = 5 // Default to 5 minutes if not set
@@ -56,9 +57,6 @@ func New(globalConfig *config.AppConfig) InstanceManager {
portRange := globalConfig.Instances.PortRange
ports := newPortAllocator(portRange[0], portRange[1])
// Initialize persistence
persistence := newInstancePersister(globalConfig.Instances.InstancesDir)
// Initialize remote manager
remote := newRemoteManager(globalConfig.Nodes, 30*time.Second)
@@ -66,7 +64,7 @@ func New(globalConfig *config.AppConfig) InstanceManager {
im := &instanceManager{
registry: registry,
ports: ports,
persistence: persistence,
db: db,
remote: remote,
globalConfig: globalConfig,
}
@@ -86,9 +84,9 @@ func New(globalConfig *config.AppConfig) InstanceManager {
return im
}
// persistInstance saves an instance using the persistence component
// persistInstance saves an instance using the persistence layer
func (im *instanceManager) persistInstance(inst *instance.Instance) error {
return im.persistence.save(inst)
return im.db.Save(inst)
}
func (im *instanceManager) Shutdown() {
@@ -116,13 +114,18 @@ func (im *instanceManager) Shutdown() {
}
wg.Wait()
fmt.Println("All instances stopped.")
// 4. Close database connection
if err := im.db.Close(); err != nil {
log.Printf("Error closing database: %v\n", err)
}
})
}
// loadInstances restores all instances from disk using the persistence component
// loadInstances restores all instances from the persistence layer
func (im *instanceManager) loadInstances() error {
// Load all instances from persistence
instances, err := im.persistence.loadAll()
instances, err := im.db.LoadAll()
if err != nil {
return fmt.Errorf("failed to load instances: %w", err)
}

View File

@@ -4,20 +4,34 @@ import (
"fmt"
"llamactl/pkg/backends"
"llamactl/pkg/config"
"llamactl/pkg/database"
"llamactl/pkg/instance"
"llamactl/pkg/manager"
"os"
"path/filepath"
"sync"
"testing"
"time"
)
func TestManager_PersistsAndLoadsInstances(t *testing.T) {
tempDir := t.TempDir()
appConfig := createTestAppConfig(tempDir)
// Use file-based database for this test since we need to persist across connections
appConfig.Database.Path = tempDir + "/test.db"
// Create instance and check file was created
manager1 := manager.New(appConfig)
// Create instance and check database was created
db1, err := database.Open(&database.Config{
Path: appConfig.Database.Path,
MaxOpenConnections: appConfig.Database.MaxOpenConnections,
MaxIdleConnections: appConfig.Database.MaxIdleConnections,
ConnMaxLifetime: appConfig.Database.ConnMaxLifetime,
})
if err != nil {
t.Fatalf("Failed to open database: %v", err)
}
if err := database.RunMigrations(db1); err != nil {
t.Fatalf("Failed to run migrations: %v", err)
}
manager1 := manager.New(appConfig, db1)
options := &instance.Options{
BackendOptions: backends.Options{
BackendType: backends.BackendTypeLlamaCpp,
@@ -28,18 +42,28 @@ func TestManager_PersistsAndLoadsInstances(t *testing.T) {
},
}
_, err := manager1.CreateInstance("test-instance", options)
_, err = manager1.CreateInstance("test-instance", options)
if err != nil {
t.Fatalf("CreateInstance failed: %v", err)
}
expectedPath := filepath.Join(tempDir, "test-instance.json")
if _, err := os.Stat(expectedPath); os.IsNotExist(err) {
t.Errorf("Expected persistence file %s to exist", expectedPath)
}
// Shutdown first manager to close database connection
manager1.Shutdown()
// Load instances from disk
manager2 := manager.New(appConfig)
// Load instances from database
db2, err := database.Open(&database.Config{
Path: appConfig.Database.Path,
MaxOpenConnections: appConfig.Database.MaxOpenConnections,
MaxIdleConnections: appConfig.Database.MaxIdleConnections,
ConnMaxLifetime: appConfig.Database.ConnMaxLifetime,
})
if err != nil {
t.Fatalf("Failed to open database: %v", err)
}
if err := database.RunMigrations(db2); err != nil {
t.Fatalf("Failed to run migrations: %v", err)
}
manager2 := manager.New(appConfig, db2)
instances, err := manager2.ListInstances()
if err != nil {
t.Fatalf("ListInstances failed: %v", err)
@@ -50,13 +74,29 @@ func TestManager_PersistsAndLoadsInstances(t *testing.T) {
if instances[0].Name != "test-instance" {
t.Errorf("Expected loaded instance name 'test-instance', got %q", instances[0].Name)
}
manager2.Shutdown()
}
func TestDeleteInstance_RemovesPersistenceFile(t *testing.T) {
func TestDeleteInstance_RemovesFromDatabase(t *testing.T) {
tempDir := t.TempDir()
appConfig := createTestAppConfig(tempDir)
mgr := manager.New(appConfig)
db, err := database.Open(&database.Config{
Path: appConfig.Database.Path,
MaxOpenConnections: appConfig.Database.MaxOpenConnections,
MaxIdleConnections: appConfig.Database.MaxIdleConnections,
ConnMaxLifetime: appConfig.Database.ConnMaxLifetime,
})
if err != nil {
t.Fatalf("Failed to open database: %v", err)
}
if err := database.RunMigrations(db); err != nil {
t.Fatalf("Failed to run migrations: %v", err)
}
mgr := manager.New(appConfig, db)
defer mgr.Shutdown()
options := &instance.Options{
BackendOptions: backends.Options{
BackendType: backends.BackendTypeLlamaCpp,
@@ -67,20 +107,33 @@ func TestDeleteInstance_RemovesPersistenceFile(t *testing.T) {
},
}
_, err := mgr.CreateInstance("test-instance", options)
_, err = mgr.CreateInstance("test-instance", options)
if err != nil {
t.Fatalf("CreateInstance failed: %v", err)
}
expectedPath := filepath.Join(tempDir, "test-instance.json")
// Verify instance exists
instances, err := mgr.ListInstances()
if err != nil {
t.Fatalf("ListInstances failed: %v", err)
}
if len(instances) != 1 {
t.Fatalf("Expected 1 instance, got %d", len(instances))
}
// Delete instance
err = mgr.DeleteInstance("test-instance")
if err != nil {
t.Fatalf("DeleteInstance failed: %v", err)
}
if _, err := os.Stat(expectedPath); !os.IsNotExist(err) {
t.Error("Expected persistence file to be deleted")
// Verify instance was deleted from database
instances, err = mgr.ListInstances()
if err != nil {
t.Fatalf("ListInstances failed: %v", err)
}
if len(instances) != 0 {
t.Errorf("Expected 0 instances after deletion, got %d", len(instances))
}
}
@@ -158,6 +211,12 @@ func createTestAppConfig(instancesDir string) *config.AppConfig {
DefaultRestartDelay: 5,
TimeoutCheckInterval: 5,
},
Database: config.DatabaseConfig{
Path: ":memory:",
MaxOpenConnections: 25,
MaxIdleConnections: 5,
ConnMaxLifetime: 5 * time.Minute,
},
LocalNode: "main",
Nodes: map[string]config.NodeConfig{},
}
@@ -166,5 +225,17 @@ func createTestAppConfig(instancesDir string) *config.AppConfig {
func createTestManager(t *testing.T) manager.InstanceManager {
tempDir := t.TempDir()
appConfig := createTestAppConfig(tempDir)
return manager.New(appConfig)
db, err := database.Open(&database.Config{
Path: appConfig.Database.Path,
MaxOpenConnections: appConfig.Database.MaxOpenConnections,
MaxIdleConnections: appConfig.Database.MaxIdleConnections,
ConnMaxLifetime: appConfig.Database.ConnMaxLifetime,
})
if err != nil {
t.Fatalf("Failed to open database: %v", err)
}
if err := database.RunMigrations(db); err != nil {
t.Fatalf("Failed to run migrations: %v", err)
}
return manager.New(appConfig, db)
}

View File

@@ -317,9 +317,9 @@ func (im *instanceManager) DeleteInstance(name string) error {
im.remote.removeInstance(name)
im.registry.remove(name)
// Delete the instance's persistence file
if err := im.persistence.delete(name); err != nil {
return fmt.Errorf("failed to delete config file for remote instance %s: %w", name, err)
// Delete the instance's persistence
if err := im.db.Delete(name); err != nil {
return fmt.Errorf("failed to delete remote instance %s: %w", name, err)
}
return nil
@@ -343,9 +343,9 @@ func (im *instanceManager) DeleteInstance(name string) error {
return fmt.Errorf("failed to remove instance from registry: %w", err)
}
// Delete persistence file
if err := im.persistence.delete(name); err != nil {
return fmt.Errorf("failed to delete config file for instance %s: %w", name, err)
// Delete from persistence
if err := im.db.Delete(name); err != nil {
return fmt.Errorf("failed to delete instance from persistence %s: %w", name, err)
}
return nil

View File

@@ -3,10 +3,12 @@ package manager_test
import (
"llamactl/pkg/backends"
"llamactl/pkg/config"
"llamactl/pkg/database"
"llamactl/pkg/instance"
"llamactl/pkg/manager"
"strings"
"testing"
"time"
)
func TestCreateInstance_FailsWithDuplicateName(t *testing.T) {
@@ -49,10 +51,28 @@ func TestCreateInstance_FailsWhenMaxInstancesReached(t *testing.T) {
MaxInstances: 1, // Very low limit for testing
TimeoutCheckInterval: 5,
},
Database: config.DatabaseConfig{
Path: ":memory:",
MaxOpenConnections: 25,
MaxIdleConnections: 5,
ConnMaxLifetime: 5 * time.Minute,
},
LocalNode: "main",
Nodes: map[string]config.NodeConfig{},
}
limitedManager := manager.New(appConfig)
db, err := database.Open(&database.Config{
Path: appConfig.Database.Path,
MaxOpenConnections: appConfig.Database.MaxOpenConnections,
MaxIdleConnections: appConfig.Database.MaxIdleConnections,
ConnMaxLifetime: appConfig.Database.ConnMaxLifetime,
})
if err != nil {
t.Fatalf("Failed to open database: %v", err)
}
if err := database.RunMigrations(db); err != nil {
t.Fatalf("Failed to run migrations: %v", err)
}
limitedManager := manager.New(appConfig, db)
options := &instance.Options{
BackendOptions: backends.Options{
@@ -63,7 +83,7 @@ func TestCreateInstance_FailsWhenMaxInstancesReached(t *testing.T) {
},
}
_, err := limitedManager.CreateInstance("instance1", options)
_, err = limitedManager.CreateInstance("instance1", options)
if err != nil {
t.Fatalf("CreateInstance 1 failed: %v", err)
}

View File

@@ -1,198 +0,0 @@
package manager
import (
"encoding/json"
"fmt"
"llamactl/pkg/instance"
"log"
"os"
"path/filepath"
"strings"
"sync"
)
// instancePersister provides atomic file-based persistence with durability guarantees.
type instancePersister struct {
mu sync.Mutex
instancesDir string
}
// newInstancePersister creates a new instance persister.
// If instancesDir is empty, persistence is disabled.
func newInstancePersister(instancesDir string) *instancePersister {
return &instancePersister{
instancesDir: instancesDir,
}
}
// Save persists an instance to disk with atomic write
func (p *instancePersister) save(inst *instance.Instance) error {
if inst == nil {
return fmt.Errorf("cannot save nil instance")
}
// Validate instance name to prevent path traversal
validatedName, err := p.validateInstanceName(inst.Name)
if err != nil {
return err
}
p.mu.Lock()
defer p.mu.Unlock()
instancePath := filepath.Join(p.instancesDir, validatedName+".json")
tempPath := instancePath + ".tmp"
// Serialize instance to JSON
jsonData, err := json.MarshalIndent(inst, "", " ")
if err != nil {
return fmt.Errorf("failed to marshal instance %s: %w", inst.Name, err)
}
// Create temporary file
tempFile, err := os.OpenFile(tempPath, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0644)
if err != nil {
return fmt.Errorf("failed to create temp file for instance %s: %w", inst.Name, err)
}
// Write data to temporary file
if _, err := tempFile.Write(jsonData); err != nil {
tempFile.Close()
os.Remove(tempPath)
return fmt.Errorf("failed to write temp file for instance %s: %w", inst.Name, err)
}
// Sync to disk before rename to ensure durability
if err := tempFile.Sync(); err != nil {
tempFile.Close()
os.Remove(tempPath)
return fmt.Errorf("failed to sync temp file for instance %s: %w", inst.Name, err)
}
// Close the file
if err := tempFile.Close(); err != nil {
os.Remove(tempPath)
return fmt.Errorf("failed to close temp file for instance %s: %w", inst.Name, err)
}
// Atomic rename (this is atomic on POSIX systems)
if err := os.Rename(tempPath, instancePath); err != nil {
os.Remove(tempPath)
return fmt.Errorf("failed to rename temp file for instance %s: %w", inst.Name, err)
}
return nil
}
// Delete removes an instance's persistence file from disk.
func (p *instancePersister) delete(name string) error {
validatedName, err := p.validateInstanceName(name)
if err != nil {
return err
}
p.mu.Lock()
defer p.mu.Unlock()
instancePath := filepath.Join(p.instancesDir, validatedName+".json")
if err := os.Remove(instancePath); err != nil {
if os.IsNotExist(err) {
// Not an error if file doesn't exist
return nil
}
return fmt.Errorf("failed to delete instance file for %s: %w", name, err)
}
return nil
}
// LoadAll loads all persisted instances from disk.
// Returns a slice of instances and any errors encountered during loading.
func (p *instancePersister) loadAll() ([]*instance.Instance, error) {
p.mu.Lock()
defer p.mu.Unlock()
// Check if instances directory exists
if _, err := os.Stat(p.instancesDir); os.IsNotExist(err) {
return nil, nil // No instances directory, return empty list
}
// Read all JSON files from instances directory
files, err := os.ReadDir(p.instancesDir)
if err != nil {
return nil, fmt.Errorf("failed to read instances directory: %w", err)
}
instances := make([]*instance.Instance, 0)
var loadErrors []string
for _, file := range files {
if file.IsDir() || !strings.HasSuffix(file.Name(), ".json") {
continue
}
instanceName := strings.TrimSuffix(file.Name(), ".json")
instancePath := filepath.Join(p.instancesDir, file.Name())
inst, err := p.loadInstanceFile(instanceName, instancePath)
if err != nil {
log.Printf("Failed to load instance %s: %v", instanceName, err)
loadErrors = append(loadErrors, fmt.Sprintf("%s: %v", instanceName, err))
continue
}
instances = append(instances, inst)
}
if len(loadErrors) > 0 {
log.Printf("Loaded %d instances with %d errors", len(instances), len(loadErrors))
} else if len(instances) > 0 {
log.Printf("Loaded %d instances from persistence", len(instances))
}
return instances, nil
}
// loadInstanceFile is an internal helper that loads a single instance file.
// Note: This assumes the mutex is already held by the caller.
func (p *instancePersister) loadInstanceFile(name, path string) (*instance.Instance, error) {
data, err := os.ReadFile(path)
if err != nil {
return nil, fmt.Errorf("failed to read instance file: %w", err)
}
var inst instance.Instance
if err := json.Unmarshal(data, &inst); err != nil {
return nil, fmt.Errorf("failed to unmarshal instance: %w", err)
}
// Validate the instance name matches the filename
if inst.Name != name {
return nil, fmt.Errorf("instance name mismatch: file=%s, instance.Name=%s", name, inst.Name)
}
return &inst, nil
}
// validateInstanceName ensures the instance name is safe for filesystem operations.
// Returns the validated name if valid, or an error if invalid.
func (p *instancePersister) validateInstanceName(name string) (string, error) {
if name == "" {
return "", fmt.Errorf("instance name cannot be empty")
}
// Check for path separators and parent directory references
// This prevents path traversal attacks
if strings.Contains(name, "/") || strings.Contains(name, "\\") || strings.Contains(name, "..") {
return "", fmt.Errorf("invalid instance name: %s (cannot contain path separators or '..')", name)
}
// Additional check: ensure the name doesn't start with a dot (hidden files)
// or contain any other suspicious characters
if strings.HasPrefix(name, ".") {
return "", fmt.Errorf("invalid instance name: %s (cannot start with '.')", name)
}
return name, nil
}