Mathis/llamactl
1
0
Fork 0
mirror of https://github.com/lordmathis/llamactl.git synced 2025-11-05 16:44:22 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #71 from lordmathis/refactor/proxy

Browse Source 
refactor: Move all proxy handling to instance package
This commit is contained in:
Matúš Námešný
2025-10-25 16:32:32 +02:00
committed by GitHub
parent 4bbf45f0b9 c038aac91b
commit 0a7420c9f9
12 changed files with 352 additions and 422 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
cmd/server/main.go
View File

@@ -58,7 +58,7 @@ func main() {
}
// Initialize the instance manager
instanceManager := manager.New(cfg.Backends, cfg.Instances, cfg.Nodes, cfg.LocalNode)
instanceManager := manager.New(&cfg)
// Create a new handler with the instance manager
handler := server.NewHandler(instanceManager, cfg)

54
pkg/instance/instance.go
View File

@@ -17,6 +17,7 @@ type Instance struct {
// Global configuration
globalInstanceSettings *config.InstancesConfig
globalBackendSettings *config.BackendConfig
globalNodesConfig map[string]config.NodeConfig
localNodeName string `json:"-"` // Name of the local node for remote detection
status *status `json:"-"`
@@ -29,7 +30,13 @@ type Instance struct {
}
// New creates a new instance with the given name, log path, options and local node name
func New(name string, globalBackendSettings *config.BackendConfig, globalInstanceSettings *config.InstancesConfig, opts *Options, localNodeName string, onStatusChange func(oldStatus, newStatus Status)) *Instance {
func New(name string, globalConfig *config.AppConfig, opts *Options, onStatusChange func(oldStatus, newStatus Status)) *Instance {
globalInstanceSettings := &globalConfig.Instances
globalBackendSettings := &globalConfig.Backends
globalNodesConfig := globalConfig.Nodes
localNodeName := globalConfig.LocalNode
// Validate and copy options
opts.validateAndApplyDefaults(name, globalInstanceSettings)
@@ -45,15 +52,21 @@ func New(name string, globalBackendSettings *config.BackendConfig, globalInstanc
options: options,
globalInstanceSettings: globalInstanceSettings,
globalBackendSettings: globalBackendSettings,
globalNodesConfig: globalNodesConfig,
localNodeName: localNodeName,
Created: time.Now().Unix(),
status: status,
}
var err error
instance.proxy, err = newProxy(instance)
if err != nil {
log.Println("Warning: Failed to create proxy for instance", instance.Name, "-", err)
}
// Only create logger, proxy, and process for local instances
if !instance.IsRemote() {
instance.logger = newLogger(name, globalInstanceSettings.LogsDir)
instance.proxy = newProxy(instance)
instance.process = newProcess(instance)
}
@@ -175,14 +188,6 @@ func (i *Instance) GetProxy() (*httputil.ReverseProxy, error) {
return nil, fmt.Errorf("instance %s has no proxy component", i.Name)
}
// Remote instances should not use local proxy - they are handled by RemoteInstanceProxy
opts := i.GetOptions()
if opts != nil && len(opts.Nodes) > 0 {
if _, isLocal := opts.Nodes[i.localNodeName]; !isLocal {
return nil, fmt.Errorf("instance %s is a remote instance and should not use local proxy", i.Name)
}
}
return i.proxy.get()
}
@@ -307,34 +312,5 @@ func (i *Instance) UnmarshalJSON(data []byte) error {
i.status = aux.Status
i.options = aux.Options
// Handle options with validation and defaults
if i.options != nil {
opts := i.options.get()
if opts != nil {
opts.validateAndApplyDefaults(i.Name, i.globalInstanceSettings)
}
}
// Initialize fields that are not serialized or may be nil
if i.status == nil {
i.status = newStatus(Stopped)
}
if i.options == nil {
i.options = newOptions(&Options{})
}
// Only create logger, proxy, and process for non-remote instances
if !i.IsRemote() {
if i.logger == nil && i.globalInstanceSettings != nil {
i.logger = newLogger(i.Name, i.globalInstanceSettings.LogsDir)
}
if i.proxy == nil {
i.proxy = newProxy(i)
}
if i.process == nil {
i.process = newProcess(i)
}
}
return nil
}

248
pkg/instance/instance_test.go
View File

@@ -11,26 +11,29 @@ import (
)
func TestNewInstance(t *testing.T) {
backendConfig := &config.BackendConfig{
LlamaCpp: config.BackendSettings{
Command: "llama-server",
Args: []string{},
globalConfig := &config.AppConfig{
Backends: config.BackendConfig{
LlamaCpp: config.BackendSettings{
Command: "llama-server",
Args: []string{},
},
MLX: config.BackendSettings{
Command: "mlx_lm.server",
Args: []string{},
},
VLLM: config.BackendSettings{
Command: "vllm",
Args: []string{"serve"},
},
},
MLX: config.BackendSettings{
Command: "mlx_lm.server",
Args: []string{},
Instances: config.InstancesConfig{
LogsDir: "/tmp/test",
DefaultAutoRestart: true,
DefaultMaxRestarts: 3,
DefaultRestartDelay: 5,
},
VLLM: config.BackendSettings{
Command: "vllm",
Args: []string{"serve"},
},
}
globalSettings := &config.InstancesConfig{
LogsDir: "/tmp/test",
DefaultAutoRestart: true,
DefaultMaxRestarts: 3,
DefaultRestartDelay: 5,
Nodes: map[string]config.NodeConfig{},
LocalNode: "main",
}
options := &instance.Options{
@@ -46,7 +49,7 @@ func TestNewInstance(t *testing.T) {
// Mock onStatusChange function
mockOnStatusChange := func(oldStatus, newStatus instance.Status) {}
inst := instance.New("test-instance", backendConfig, globalSettings, options, "main", mockOnStatusChange)
inst := instance.New("test-instance", globalConfig, options, mockOnStatusChange)
if inst.Name != "test-instance" {
t.Errorf("Expected name 'test-instance', got %q", inst.Name)
@@ -79,8 +82,8 @@ func TestNewInstance(t *testing.T) {
autoRestart := false
maxRestarts := 10
optionsWithOverrides := &instance.Options{
AutoRestart: &autoRestart,
MaxRestarts: &maxRestarts,
AutoRestart: &autoRestart,
MaxRestarts: &maxRestarts,
BackendOptions: backends.Options{
BackendType: backends.BackendTypeLlamaCpp,
LlamaServerOptions: &backends.LlamaServerOptions{
@@ -89,7 +92,7 @@ func TestNewInstance(t *testing.T) {
},
}
inst2 := instance.New("test-override", backendConfig, globalSettings, optionsWithOverrides, "main", mockOnStatusChange)
inst2 := instance.New("test-override", globalConfig, optionsWithOverrides, mockOnStatusChange)
opts2 := inst2.GetOptions()
if opts2.AutoRestart == nil || *opts2.AutoRestart {
@@ -101,26 +104,29 @@ func TestNewInstance(t *testing.T) {
}
func TestSetOptions(t *testing.T) {
backendConfig := &config.BackendConfig{
LlamaCpp: config.BackendSettings{
Command: "llama-server",
Args: []string{},
globalConfig := &config.AppConfig{
Backends: config.BackendConfig{
LlamaCpp: config.BackendSettings{
Command: "llama-server",
Args: []string{},
},
MLX: config.BackendSettings{
Command: "mlx_lm.server",
Args: []string{},
},
VLLM: config.BackendSettings{
Command: "vllm",
Args: []string{"serve"},
},
},
MLX: config.BackendSettings{
Command: "mlx_lm.server",
Args: []string{},
Instances: config.InstancesConfig{
LogsDir: "/tmp/test",
DefaultAutoRestart: true,
DefaultMaxRestarts: 3,
DefaultRestartDelay: 5,
},
VLLM: config.BackendSettings{
Command: "vllm",
Args: []string{"serve"},
},
}
globalSettings := &config.InstancesConfig{
LogsDir: "/tmp/test",
DefaultAutoRestart: true,
DefaultMaxRestarts: 3,
DefaultRestartDelay: 5,
Nodes: map[string]config.NodeConfig{},
LocalNode: "main",
}
initialOptions := &instance.Options{
@@ -136,7 +142,7 @@ func TestSetOptions(t *testing.T) {
// Mock onStatusChange function
mockOnStatusChange := func(oldStatus, newStatus instance.Status) {}
inst := instance.New("test-instance", backendConfig, globalSettings, initialOptions, "main", mockOnStatusChange)
inst := instance.New("test-instance", globalConfig, initialOptions, mockOnStatusChange)
// Update options
newOptions := &instance.Options{
@@ -166,23 +172,26 @@ func TestSetOptions(t *testing.T) {
}
func TestGetProxy(t *testing.T) {
backendConfig := &config.BackendConfig{
LlamaCpp: config.BackendSettings{
Command: "llama-server",
Args: []string{},
globalConfig := &config.AppConfig{
Backends: config.BackendConfig{
LlamaCpp: config.BackendSettings{
Command: "llama-server",
Args: []string{},
},
MLX: config.BackendSettings{
Command: "mlx_lm.server",
Args: []string{},
},
VLLM: config.BackendSettings{
Command: "vllm",
Args: []string{"serve"},
},
},
MLX: config.BackendSettings{
Command: "mlx_lm.server",
Args: []string{},
Instances: config.InstancesConfig{
LogsDir: "/tmp/test",
},
VLLM: config.BackendSettings{
Command: "vllm",
Args: []string{"serve"},
},
}
globalSettings := &config.InstancesConfig{
LogsDir: "/tmp/test",
Nodes: map[string]config.NodeConfig{},
LocalNode: "main",
}
options := &instance.Options{
@@ -199,7 +208,7 @@ func TestGetProxy(t *testing.T) {
// Mock onStatusChange function
mockOnStatusChange := func(oldStatus, newStatus instance.Status) {}
inst := instance.New("test-instance", backendConfig, globalSettings, options, "main", mockOnStatusChange)
inst := instance.New("test-instance", globalConfig, options, mockOnStatusChange)
// Get proxy for the first time
proxy1, err := inst.GetProxy()
@@ -221,10 +230,14 @@ func TestGetProxy(t *testing.T) {
}
func TestMarshalJSON(t *testing.T) {
backendConfig := &config.BackendConfig{
LlamaCpp: config.BackendSettings{Command: "llama-server"},
globalConfig := &config.AppConfig{
Backends: config.BackendConfig{
LlamaCpp: config.BackendSettings{Command: "llama-server"},
},
Instances: config.InstancesConfig{LogsDir: "/tmp/test"},
Nodes: map[string]config.NodeConfig{},
LocalNode: "main",
}
globalSettings := &config.InstancesConfig{LogsDir: "/tmp/test"}
options := &instance.Options{
BackendOptions: backends.Options{
BackendType: backends.BackendTypeLlamaCpp,
@@ -235,7 +248,7 @@ func TestMarshalJSON(t *testing.T) {
},
}
inst := instance.New("test-instance", backendConfig, globalSettings, options, "main", nil)
inst := instance.New("test-instance", globalConfig, options, nil)
data, err := json.Marshal(inst)
if err != nil {
@@ -342,23 +355,26 @@ func TestCreateOptionsValidation(t *testing.T) {
},
}
backendConfig := &config.BackendConfig{
LlamaCpp: config.BackendSettings{
Command: "llama-server",
Args: []string{},
globalConfig := &config.AppConfig{
Backends: config.BackendConfig{
LlamaCpp: config.BackendSettings{
Command: "llama-server",
Args: []string{},
},
MLX: config.BackendSettings{
Command: "mlx_lm.server",
Args: []string{},
},
VLLM: config.BackendSettings{
Command: "vllm",
Args: []string{"serve"},
},
},
MLX: config.BackendSettings{
Command: "mlx_lm.server",
Args: []string{},
Instances: config.InstancesConfig{
LogsDir: "/tmp/test",
},
VLLM: config.BackendSettings{
Command: "vllm",
Args: []string{"serve"},
},
}
globalSettings := &config.InstancesConfig{
LogsDir: "/tmp/test",
Nodes: map[string]config.NodeConfig{},
LocalNode: "main",
}
for _, tt := range tests {
@@ -377,7 +393,7 @@ func TestCreateOptionsValidation(t *testing.T) {
// Mock onStatusChange function
mockOnStatusChange := func(oldStatus, newStatus instance.Status) {}
instance := instance.New("test", backendConfig, globalSettings, options, "main", mockOnStatusChange)
instance := instance.New("test", globalConfig, options, mockOnStatusChange)
opts := instance.GetOptions()
if opts.MaxRestarts == nil {
@@ -396,10 +412,14 @@ func TestCreateOptionsValidation(t *testing.T) {
}
func TestStatusChangeCallback(t *testing.T) {
backendConfig := &config.BackendConfig{
LlamaCpp: config.BackendSettings{Command: "llama-server"},
globalConfig := &config.AppConfig{
Backends: config.BackendConfig{
LlamaCpp: config.BackendSettings{Command: "llama-server"},
},
Instances: config.InstancesConfig{LogsDir: "/tmp/test"},
Nodes: map[string]config.NodeConfig{},
LocalNode: "main",
}
globalSettings := &config.InstancesConfig{LogsDir: "/tmp/test"}
options := &instance.Options{
BackendOptions: backends.Options{
BackendType: backends.BackendTypeLlamaCpp,
@@ -418,7 +438,7 @@ func TestStatusChangeCallback(t *testing.T) {
callbackCalled = true
}
inst := instance.New("test", backendConfig, globalSettings, options, "main", onStatusChange)
inst := instance.New("test", globalConfig, options, onStatusChange)
inst.SetStatus(instance.Running)
@@ -434,10 +454,14 @@ func TestStatusChangeCallback(t *testing.T) {
}
func TestSetOptions_NodesPreserved(t *testing.T) {
backendConfig := &config.BackendConfig{
LlamaCpp: config.BackendSettings{Command: "llama-server"},
globalConfig := &config.AppConfig{
Backends: config.BackendConfig{
LlamaCpp: config.BackendSettings{Command: "llama-server"},
},
Instances: config.InstancesConfig{LogsDir: "/tmp/test"},
Nodes: map[string]config.NodeConfig{},
LocalNode: "main",
}
globalSettings := &config.InstancesConfig{LogsDir: "/tmp/test"}
tests := []struct {
name string
@@ -477,7 +501,7 @@ func TestSetOptions_NodesPreserved(t *testing.T) {
},
}
inst := instance.New("test", backendConfig, globalSettings, options, "main", nil)
inst := instance.New("test", globalConfig, options, nil)
// Attempt to update nodes (should be ignored)
updateOptions := &instance.Options{
@@ -512,10 +536,14 @@ func TestSetOptions_NodesPreserved(t *testing.T) {
}
func TestProcessErrorCases(t *testing.T) {
backendConfig := &config.BackendConfig{
LlamaCpp: config.BackendSettings{Command: "llama-server"},
globalConfig := &config.AppConfig{
Backends: config.BackendConfig{
LlamaCpp: config.BackendSettings{Command: "llama-server"},
},
Instances: config.InstancesConfig{LogsDir: "/tmp/test"},
Nodes: map[string]config.NodeConfig{},
LocalNode: "main",
}
globalSettings := &config.InstancesConfig{LogsDir: "/tmp/test"}
options := &instance.Options{
BackendOptions: backends.Options{
BackendType: backends.BackendTypeLlamaCpp,
@@ -525,7 +553,7 @@ func TestProcessErrorCases(t *testing.T) {
},
}
inst := instance.New("test", backendConfig, globalSettings, options, "main", nil)
inst := instance.New("test", globalConfig, options, nil)
// Stop when not running should return error
err := inst.Stop()
@@ -544,10 +572,16 @@ func TestProcessErrorCases(t *testing.T) {
}
func TestRemoteInstanceOperations(t *testing.T) {
backendConfig := &config.BackendConfig{
LlamaCpp: config.BackendSettings{Command: "llama-server"},
globalConfig := &config.AppConfig{
Backends: config.BackendConfig{
LlamaCpp: config.BackendSettings{Command: "llama-server"},
},
Instances: config.InstancesConfig{LogsDir: "/tmp/test"},
Nodes: map[string]config.NodeConfig{
"remote-node": {Address: "http://remote-node:8080"},
},
LocalNode: "main",
}
globalSettings := &config.InstancesConfig{LogsDir: "/tmp/test"}
options := &instance.Options{
Nodes: map[string]struct{}{"remote-node": {}}, // Remote instance
BackendOptions: backends.Options{
@@ -558,7 +592,7 @@ func TestRemoteInstanceOperations(t *testing.T) {
},
}
inst := instance.New("remote-test", backendConfig, globalSettings, options, "main", nil)
inst := instance.New("remote-test", globalConfig, options, nil)
if !inst.IsRemote() {
t.Error("Expected instance to be remote")
@@ -580,8 +614,8 @@ func TestRemoteInstanceOperations(t *testing.T) {
}
// GetProxy should fail for remote instance
if _, err := inst.GetProxy(); err == nil {
t.Error("Expected error when getting proxy for remote instance")
if _, err := inst.GetProxy(); err != nil {
t.Error("Expected no error when getting proxy for remote instance")
}
// GetLogs should fail for remote instance
@@ -591,14 +625,18 @@ func TestRemoteInstanceOperations(t *testing.T) {
}
func TestIdleTimeout(t *testing.T) {
backendConfig := &config.BackendConfig{
LlamaCpp: config.BackendSettings{Command: "llama-server"},
globalConfig := &config.AppConfig{
Backends: config.BackendConfig{
LlamaCpp: config.BackendSettings{Command: "llama-server"},
},
Instances: config.InstancesConfig{LogsDir: "/tmp/test"},
Nodes: map[string]config.NodeConfig{},
LocalNode: "main",
}
globalSettings := &config.InstancesConfig{LogsDir: "/tmp/test"}
t.Run("not running never times out", func(t *testing.T) {
timeout := 1
inst := instance.New("test", backendConfig, globalSettings, &instance.Options{
inst := instance.New("test", globalConfig, &instance.Options{
IdleTimeout: &timeout,
BackendOptions: backends.Options{
BackendType: backends.BackendTypeLlamaCpp,
@@ -606,7 +644,7 @@ func TestIdleTimeout(t *testing.T) {
Model: "/path/to/model.gguf",
},
},
}, "main", nil)
}, nil)
if inst.ShouldTimeout() {
t.Error("Non-running instance should never timeout")
@@ -614,7 +652,7 @@ func TestIdleTimeout(t *testing.T) {
})
t.Run("no timeout configured", func(t *testing.T) {
inst := instance.New("test", backendConfig, globalSettings, &instance.Options{
inst := instance.New("test", globalConfig, &instance.Options{
IdleTimeout: nil, // No timeout
BackendOptions: backends.Options{
BackendType: backends.BackendTypeLlamaCpp,
@@ -622,7 +660,7 @@ func TestIdleTimeout(t *testing.T) {
Model: "/path/to/model.gguf",
},
},
}, "main", nil)
}, nil)
inst.SetStatus(instance.Running)
if inst.ShouldTimeout() {
@@ -632,15 +670,17 @@ func TestIdleTimeout(t *testing.T) {
t.Run("timeout exceeded", func(t *testing.T) {
timeout := 1 // 1 minute
inst := instance.New("test", backendConfig, globalSettings, &instance.Options{
inst := instance.New("test", globalConfig, &instance.Options{
IdleTimeout: &timeout,
BackendOptions: backends.Options{
BackendType: backends.BackendTypeLlamaCpp,
LlamaServerOptions: &backends.LlamaServerOptions{
Model: "/path/to/model.gguf",
Host: "localhost",
Port: 8080,
},
},
}, "main", nil)
}, nil)
inst.SetStatus(instance.Running)
// Use mock time provider

136
pkg/instance/proxy.go
View File

@@ -26,20 +26,78 @@ func (realTimeProvider) Now() time.Time {
type proxy struct {
instance *Instance
mu sync.RWMutex
proxy *httputil.ReverseProxy
proxyOnce sync.Once
proxyErr error
targetURL *url.URL
apiKey string // For remote instances
responseHeaders map[string]string
mu sync.RWMutex
proxy *httputil.ReverseProxy
proxyOnce sync.Once
proxyErr error
lastRequestTime atomic.Int64
timeProvider TimeProvider
}
// newProxy creates a new Proxy for the given instance
func newProxy(instance *Instance) *proxy {
return &proxy{
func newProxy(instance *Instance) (*proxy, error) {
p := &proxy{
instance: instance,
timeProvider: realTimeProvider{},
}
var err error
options := instance.GetOptions()
if options == nil {
return nil, fmt.Errorf("instance %s has no options set", instance.Name)
}
if instance.IsRemote() {
// Take the first remote node as the target for now
var nodeName string
for node := range options.Nodes {
nodeName = node
break
}
if nodeName == "" {
return nil, fmt.Errorf("instance %s has no remote nodes defined", p.instance.Name)
}
node, ok := p.instance.globalNodesConfig[nodeName]
if !ok {
return nil, fmt.Errorf("remote node %s is not defined", nodeName)
}
p.targetURL, err = url.Parse(node.Address)
if err != nil {
return nil, fmt.Errorf("failed to parse target URL for remote instance %s: %w", p.instance.Name, err)
}
p.apiKey = node.APIKey
} else {
// Get host/port from process
host := p.instance.options.GetHost()
port := p.instance.options.GetPort()
if port == 0 {
return nil, fmt.Errorf("instance %s has no port assigned", p.instance.Name)
}
p.targetURL, err = url.Parse(fmt.Sprintf("http://%s:%d", host, port))
if err != nil {
return nil, fmt.Errorf("failed to parse target URL for instance %s: %w", p.instance.Name, err)
}
// Get response headers from backend config
p.responseHeaders = options.BackendOptions.GetResponseHeaders(p.instance.globalBackendSettings)
}
return p, nil
}
// get returns the reverse proxy for this instance, creating it if needed.
@@ -56,46 +114,40 @@ func (p *proxy) get() (*httputil.ReverseProxy, error) {
// build creates the reverse proxy based on instance options
func (p *proxy) build() (*httputil.ReverseProxy, error) {
options := p.instance.GetOptions()
if options == nil {
return nil, fmt.Errorf("instance %s has no options set", p.instance.Name)
}
// Remote instances should not use local proxy - they are handled by RemoteInstanceProxy
if _, isLocal := options.Nodes[p.instance.localNodeName]; !isLocal {
return nil, fmt.Errorf("instance %s is a remote instance and should not use local proxy", p.instance.Name)
}
proxy := httputil.NewSingleHostReverseProxy(p.targetURL)
// Get host/port from process
host := p.instance.options.GetHost()
port := p.instance.options.GetPort()
if port == 0 {
return nil, fmt.Errorf("instance %s has no port assigned", p.instance.Name)
}
targetURL, err := url.Parse(fmt.Sprintf("http://%s:%d", host, port))
if err != nil {
return nil, fmt.Errorf("failed to parse target URL for instance %s: %w", p.instance.Name, err)
}
// Modify the request before sending it to the backend
originalDirector := proxy.Director
proxy.Director = func(req *http.Request) {
originalDirector(req)
proxy := httputil.NewSingleHostReverseProxy(targetURL)
// Get response headers from backend config
responseHeaders := options.BackendOptions.GetResponseHeaders(p.instance.globalBackendSettings)
proxy.ModifyResponse = func(resp *http.Response) error {
// Remove CORS headers from backend response to avoid conflicts
// llamactl will add its own CORS headers
resp.Header.Del("Access-Control-Allow-Origin")
resp.Header.Del("Access-Control-Allow-Methods")
resp.Header.Del("Access-Control-Allow-Headers")
resp.Header.Del("Access-Control-Allow-Credentials")
resp.Header.Del("Access-Control-Max-Age")
resp.Header.Del("Access-Control-Expose-Headers")
for key, value := range responseHeaders {
resp.Header.Set(key, value)
// Add API key header for remote instances
if p.instance.IsRemote() && p.apiKey != "" {
req.Header.Set("Authorization", "Bearer "+p.apiKey)
}
// Update last request time
p.updateLastRequestTime()
}
if !p.instance.IsRemote() {
// Add custom headers to the request
proxy.ModifyResponse = func(resp *http.Response) error {
// Remove CORS headers from backend response to avoid conflicts
// llamactl will add its own CORS headers
resp.Header.Del("Access-Control-Allow-Origin")
resp.Header.Del("Access-Control-Allow-Methods")
resp.Header.Del("Access-Control-Allow-Headers")
resp.Header.Del("Access-Control-Allow-Credentials")
resp.Header.Del("Access-Control-Max-Age")
resp.Header.Del("Access-Control-Expose-Headers")
for key, value := range p.responseHeaders {
resp.Header.Set(key, value)
}
return nil
}
return nil
}
return proxy, nil

35
pkg/manager/manager.go
View File

@@ -35,9 +35,7 @@ type instanceManager struct {
lifecycle *lifecycleManager
// Configuration
instancesConfig config.InstancesConfig
backendsConfig config.BackendConfig
localNodeName string // Name of the local node
globalConfig *config.AppConfig
// Synchronization
instanceLocks sync.Map // map[string]*sync.Mutex - per-instance locks for concurrent operations
@@ -45,43 +43,42 @@ type instanceManager struct {
}
// New creates a new instance of InstanceManager.
func New(backendsConfig config.BackendConfig, instancesConfig config.InstancesConfig, nodesConfig map[string]config.NodeConfig, localNodeName string) InstanceManager {
if instancesConfig.TimeoutCheckInterval <= 0 {
instancesConfig.TimeoutCheckInterval = 5 // Default to 5 minutes if not set
func New(globalConfig *config.AppConfig) InstanceManager {
if globalConfig.Instances.TimeoutCheckInterval <= 0 {
globalConfig.Instances.TimeoutCheckInterval = 5 // Default to 5 minutes if not set
}
// Initialize components
registry := newInstanceRegistry()
// Initialize port allocator
portRange := instancesConfig.PortRange
portRange := globalConfig.Instances.PortRange
ports, err := newPortAllocator(portRange[0], portRange[1])
if err != nil {
log.Fatalf("Failed to create port allocator: %v", err)
}
// Initialize persistence
persistence, err := newInstancePersister(instancesConfig.InstancesDir)
persistence, err := newInstancePersister(globalConfig.Instances.InstancesDir)
if err != nil {
log.Fatalf("Failed to create instance persister: %v", err)
}
// Initialize remote manager
remote := newRemoteManager(nodesConfig, 30*time.Second)
remote := newRemoteManager(globalConfig.Nodes, 30*time.Second)
// Create manager instance
im := &instanceManager{
registry: registry,
ports: ports,
persistence: persistence,
remote: remote,
instancesConfig: instancesConfig,
backendsConfig: backendsConfig,
localNodeName: localNodeName,
registry: registry,
ports: ports,
persistence: persistence,
remote: remote,
globalConfig: globalConfig,
}
// Initialize lifecycle manager (needs reference to manager for Stop/Evict operations)
checkInterval := time.Duration(instancesConfig.TimeoutCheckInterval) * time.Minute
checkInterval := time.Duration(globalConfig.Instances.TimeoutCheckInterval) * time.Minute
im.lifecycle = newLifecycleManager(registry, im, checkInterval, true)
// Load existing instances from disk
@@ -165,7 +162,7 @@ func (im *instanceManager) loadInstance(persistedInst *instance.Instance) error
var isRemote bool
var nodeName string
if options != nil {
if _, isLocal := options.Nodes[im.localNodeName]; !isLocal && len(options.Nodes) > 0 {
if _, isLocal := options.Nodes[im.globalConfig.LocalNode]; !isLocal && len(options.Nodes) > 0 {
// Get the first node from the set
for node := range options.Nodes {
nodeName = node
@@ -184,7 +181,7 @@ func (im *instanceManager) loadInstance(persistedInst *instance.Instance) error
}
// Create new inst using NewInstance (handles validation, defaults, setup)
inst := instance.New(name, &im.backendsConfig, &im.instancesConfig, options, im.localNodeName, statusCallback)
inst := instance.New(name, im.globalConfig, options, statusCallback)
// Restore persisted fields that NewInstance doesn't set
inst.Created = persistedInst.Created

80
pkg/manager/manager_test.go
View File

@@ -14,11 +14,10 @@ import (
func TestManager_PersistsAndLoadsInstances(t *testing.T) {
tempDir := t.TempDir()
cfg := createPersistenceConfig(tempDir)
backendConfig := createBackendConfig()
appConfig := createTestAppConfig(tempDir)
// Create instance and check file was created
manager1 := manager.New(backendConfig, cfg, map[string]config.NodeConfig{}, "main")
manager1 := manager.New(appConfig)
options := &instance.Options{
BackendOptions: backends.Options{
BackendType: backends.BackendTypeLlamaCpp,
@@ -40,7 +39,7 @@ func TestManager_PersistsAndLoadsInstances(t *testing.T) {
}
// Load instances from disk
manager2 := manager.New(backendConfig, cfg, map[string]config.NodeConfig{}, "main")
manager2 := manager.New(appConfig)
instances, err := manager2.ListInstances()
if err != nil {
t.Fatalf("ListInstances failed: %v", err)
@@ -55,10 +54,9 @@ func TestManager_PersistsAndLoadsInstances(t *testing.T) {
func TestDeleteInstance_RemovesPersistenceFile(t *testing.T) {
tempDir := t.TempDir()
cfg := createPersistenceConfig(tempDir)
backendConfig := createBackendConfig()
appConfig := createTestAppConfig(tempDir)
mgr := manager.New(backendConfig, cfg, map[string]config.NodeConfig{}, "main")
mgr := manager.New(appConfig)
options := &instance.Options{
BackendOptions: backends.Options{
BackendType: backends.BackendTypeLlamaCpp,
@@ -135,39 +133,57 @@ func TestConcurrentAccess(t *testing.T) {
}
// Helper functions for test configuration
func createBackendConfig() config.BackendConfig {
func createTestAppConfig(instancesDir string) *config.AppConfig {
// Use 'sleep' as a test command instead of 'llama-server'
// This allows tests to run in CI environments without requiring actual LLM binaries
// The sleep command will be invoked with model paths and other args, which it ignores
return config.BackendConfig{
LlamaCpp: config.BackendSettings{
Command: "sleep",
return &config.AppConfig{
Backends: config.BackendConfig{
LlamaCpp: config.BackendSettings{
Command: "sleep",
},
MLX: config.BackendSettings{
Command: "sleep",
},
},
MLX: config.BackendSettings{
Command: "sleep",
Instances: config.InstancesConfig{
PortRange: [2]int{8000, 9000},
InstancesDir: instancesDir,
LogsDir: instancesDir,
MaxInstances: 10,
MaxRunningInstances: 10,
DefaultAutoRestart: true,
DefaultMaxRestarts: 3,
DefaultRestartDelay: 5,
TimeoutCheckInterval: 5,
},
}
}
func createPersistenceConfig(dir string) config.InstancesConfig {
return config.InstancesConfig{
PortRange: [2]int{8000, 9000},
InstancesDir: dir,
MaxInstances: 10,
TimeoutCheckInterval: 5,
LocalNode: "main",
Nodes: map[string]config.NodeConfig{},
}
}
func createTestManager() manager.InstanceManager {
cfg := config.InstancesConfig{
PortRange: [2]int{8000, 9000},
LogsDir: "/tmp/test",
MaxInstances: 10,
MaxRunningInstances: 10,
DefaultAutoRestart: true,
DefaultMaxRestarts: 3,
DefaultRestartDelay: 5,
TimeoutCheckInterval: 5,
appConfig := &config.AppConfig{
Backends: config.BackendConfig{
LlamaCpp: config.BackendSettings{
Command: "sleep",
},
MLX: config.BackendSettings{
Command: "sleep",
},
},
Instances: config.InstancesConfig{
PortRange: [2]int{8000, 9000},
LogsDir: "/tmp/test",
MaxInstances: 10,
MaxRunningInstances: 10,
DefaultAutoRestart: true,
DefaultMaxRestarts: 3,
DefaultRestartDelay: 5,
TimeoutCheckInterval: 5,
},
LocalNode: "main",
Nodes: map[string]config.NodeConfig{},
}
return manager.New(createBackendConfig(), cfg, map[string]config.NodeConfig{}, "main")
return manager.New(appConfig)
}

16
pkg/manager/operations.go
View File

@@ -68,7 +68,7 @@ func (im *instanceManager) CreateInstance(name string, options *instance.Options
}
// Check if this is a remote instance (local node not in the Nodes set)
if _, isLocal := options.Nodes[im.localNodeName]; !isLocal && len(options.Nodes) > 0 {
if _, isLocal := options.Nodes[im.globalConfig.LocalNode]; !isLocal && len(options.Nodes) > 0 {
// Get the first node from the set
var nodeName string
for node := range options.Nodes {
@@ -94,7 +94,7 @@ func (im *instanceManager) CreateInstance(name string, options *instance.Options
// Create a local stub that preserves the Nodes field for tracking
// We keep the original options (with Nodes) so IsRemote() works correctly
inst := instance.New(name, &im.backendsConfig, &im.instancesConfig, options, im.localNodeName, nil)
inst := instance.New(name, im.globalConfig, options, nil)
// Update the local stub with all remote data (preserving Nodes)
im.updateLocalInstanceFromRemote(inst, remoteInst)
@@ -129,8 +129,8 @@ func (im *instanceManager) CreateInstance(name string, options *instance.Options
}
}
localInstanceCount := totalInstances - remoteCount
if localInstanceCount >= im.instancesConfig.MaxInstances && im.instancesConfig.MaxInstances != -1 {
return nil, fmt.Errorf("maximum number of instances (%d) reached", im.instancesConfig.MaxInstances)
if localInstanceCount >= im.globalConfig.Instances.MaxInstances && im.globalConfig.Instances.MaxInstances != -1 {
return nil, fmt.Errorf("maximum number of instances (%d) reached", im.globalConfig.Instances.MaxInstances)
}
// Assign and validate port for backend-specific options
@@ -155,7 +155,7 @@ func (im *instanceManager) CreateInstance(name string, options *instance.Options
im.onStatusChange(name, oldStatus, newStatus)
}
inst := instance.New(name, &im.backendsConfig, &im.instancesConfig, options, im.localNodeName, statusCallback)
inst := instance.New(name, im.globalConfig, options, statusCallback)
// Add to registry
if err := im.registry.add(inst); err != nil {
@@ -384,7 +384,7 @@ func (im *instanceManager) StartInstance(name string) (*instance.Instance, error
// Check max running instances limit for local instances only
if im.IsMaxRunningInstancesReached() {
return nil, MaxRunningInstancesError(fmt.Errorf("maximum number of running instances (%d) reached", im.instancesConfig.MaxRunningInstances))
return nil, MaxRunningInstancesError(fmt.Errorf("maximum number of running instances (%d) reached", im.globalConfig.Instances.MaxRunningInstances))
}
if err := inst.Start(); err != nil {
@@ -400,7 +400,7 @@ func (im *instanceManager) StartInstance(name string) (*instance.Instance, error
}
func (im *instanceManager) IsMaxRunningInstancesReached() bool {
if im.instancesConfig.MaxRunningInstances == -1 {
if im.globalConfig.Instances.MaxRunningInstances == -1 {
return false
}
@@ -412,7 +412,7 @@ func (im *instanceManager) IsMaxRunningInstancesReached() bool {
}
}
return localRunningCount >= im.instancesConfig.MaxRunningInstances
return localRunningCount >= im.globalConfig.Instances.MaxRunningInstances
}
// StopInstance stops a running instance and returns it.

22
pkg/manager/operations_test.go
View File

@@ -36,17 +36,21 @@ func TestCreateInstance_FailsWithDuplicateName(t *testing.T) {
}
func TestCreateInstance_FailsWhenMaxInstancesReached(t *testing.T) {
backendConfig := config.BackendConfig{
LlamaCpp: config.BackendSettings{
Command: "llama-server",
appConfig := &config.AppConfig{
Backends: config.BackendConfig{
LlamaCpp: config.BackendSettings{
Command: "llama-server",
},
},
Instances: config.InstancesConfig{
PortRange: [2]int{8000, 9000},
MaxInstances: 1, // Very low limit for testing
TimeoutCheckInterval: 5,
},
LocalNode: "main",
Nodes: map[string]config.NodeConfig{},
}
cfg := config.InstancesConfig{
PortRange: [2]int{8000, 9000},
MaxInstances: 1, // Very low limit for testing
TimeoutCheckInterval: 5,
}
limitedManager := manager.New(backendConfig, cfg, map[string]config.NodeConfig{}, "main")
limitedManager := manager.New(appConfig)
options := &instance.Options{
BackendOptions: backends.Options{

5
pkg/server/handlers.go
View File

@@ -4,8 +4,6 @@ import (
"llamactl/pkg/config"
"llamactl/pkg/manager"
"net/http"
"net/http/httputil"
"sync"
"time"
)
@@ -13,8 +11,6 @@ type Handler struct {
InstanceManager manager.InstanceManager
cfg config.AppConfig
httpClient *http.Client
remoteProxies map[string]*httputil.ReverseProxy // Cache of remote proxies by instance name
remoteProxiesMu sync.RWMutex
}
func NewHandler(im manager.InstanceManager, cfg config.AppConfig) *Handler {
@@ -24,6 +20,5 @@ func NewHandler(im manager.InstanceManager, cfg config.AppConfig) *Handler {
httpClient: &http.Client{
Timeout: 30 * time.Second,
},
remoteProxies: make(map[string]*httputil.ReverseProxy),
}
}

13
pkg/server/handlers_backends.go
View File

@@ -49,7 +49,7 @@ func (h *Handler) LlamaCppProxy(onDemandStart bool) http.HandlerFunc {
return
}
if !inst.IsRunning() {
if !inst.IsRemote() && !inst.IsRunning() {
if !(onDemandStart && options.OnDemandStart != nil && *options.OnDemandStart) {
http.Error(w, "Instance is not running", http.StatusServiceUnavailable)
@@ -88,12 +88,11 @@ func (h *Handler) LlamaCppProxy(onDemandStart bool) http.HandlerFunc {
return
}
// Strip the "/llama-cpp/<name>" prefix from the request URL
prefix := fmt.Sprintf("/llama-cpp/%s", validatedName)
r.URL.Path = strings.TrimPrefix(r.URL.Path, prefix)
// Update the last request time for the instance
inst.UpdateLastRequestTime()
if !inst.IsRemote() {
// Strip the "/llama-cpp/<name>" prefix from the request URL
prefix := fmt.Sprintf("/llama-cpp/%s", validatedName)
r.URL.Path = strings.TrimPrefix(r.URL.Path, prefix)
}
proxy.ServeHTTP(w, r)
}

83
pkg/server/handlers_instances.go
View File

@@ -7,8 +7,6 @@ import (
"llamactl/pkg/manager"
"llamactl/pkg/validation"
"net/http"
"net/http/httputil"
"net/url"
"strconv"
"strings"
@@ -375,12 +373,6 @@ func (h *Handler) ProxyToInstance() http.HandlerFunc {
return
}
// Check if this is a remote instance
if inst.IsRemote() {
h.RemoteInstanceProxy(w, r, validatedName, inst)
return
}
if !inst.IsRunning() {
http.Error(w, "Instance is not running", http.StatusServiceUnavailable)
return
@@ -393,12 +385,12 @@ func (h *Handler) ProxyToInstance() http.HandlerFunc {
return
}
// Strip the "/api/v1/instances/<name>/proxy" prefix from the request URL
prefix := fmt.Sprintf("/api/v1/instances/%s/proxy", validatedName)
r.URL.Path = strings.TrimPrefix(r.URL.Path, prefix)
// Update the last request time for the instance
inst.UpdateLastRequestTime()
// Check if this is a remote instance
if !inst.IsRemote() {
// Strip the "/api/v1/instances/<name>/proxy" prefix from the request URL
prefix := fmt.Sprintf("/api/v1/instances/%s/proxy", validatedName)
r.URL.Path = strings.TrimPrefix(r.URL.Path, prefix)
}
// Set forwarded headers
r.Header.Set("X-Forwarded-Host", r.Header.Get("Host"))
@@ -408,66 +400,3 @@ func (h *Handler) ProxyToInstance() http.HandlerFunc {
proxy.ServeHTTP(w, r)
}
}
// RemoteInstanceProxy proxies requests to a remote instance
func (h *Handler) RemoteInstanceProxy(w http.ResponseWriter, r *http.Request, name string, inst *instance.Instance) {
// Get the node name from instance options
options := inst.GetOptions()
if options == nil {
http.Error(w, "Instance has no options configured", http.StatusInternalServerError)
return
}
// Get the first node from the set
var nodeName string
for node := range options.Nodes {
nodeName = node
break
}
if nodeName == "" {
http.Error(w, "Instance has no node configured", http.StatusInternalServerError)
return
}
// Check if we have a cached proxy for this node
h.remoteProxiesMu.RLock()
proxy, exists := h.remoteProxies[nodeName]
h.remoteProxiesMu.RUnlock()
if !exists {
// Find node configuration
nodeConfig, exists := h.cfg.Nodes[nodeName]
if !exists {
http.Error(w, fmt.Sprintf("Node %s not found", nodeName), http.StatusInternalServerError)
return
}
// Create reverse proxy to remote node
targetURL, err := url.Parse(nodeConfig.Address)
if err != nil {
http.Error(w, "Failed to parse node address: "+err.Error(), http.StatusInternalServerError)
return
}
proxy = httputil.NewSingleHostReverseProxy(targetURL)
// Modify request before forwarding
originalDirector := proxy.Director
apiKey := nodeConfig.APIKey // Capture for closure
proxy.Director = func(req *http.Request) {
originalDirector(req)
// Add API key if configured
if apiKey != "" {
req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", apiKey))
}
}
// Cache the proxy by node name
h.remoteProxiesMu.Lock()
h.remoteProxies[nodeName] = proxy
h.remoteProxiesMu.Unlock()
}
// Forward the request using the cached proxy
proxy.ServeHTTP(w, r)
}

80
pkg/server/handlers_openai.go
View File

@@ -3,13 +3,9 @@ package server
import (
"bytes"
"encoding/json"
"fmt"
"io"
"llamactl/pkg/instance"
"llamactl/pkg/validation"
"net/http"
"net/http/httputil"
"net/url"
)
// OpenAIListInstances godoc
@@ -100,15 +96,7 @@ func (h *Handler) OpenAIProxy() http.HandlerFunc {
return
}
// Check if this is a remote instance
if inst.IsRemote() {
// Restore the body for the remote proxy
r.Body = io.NopCloser(bytes.NewReader(bodyBytes))
h.RemoteOpenAIProxy(w, r, validatedName, inst)
return
}
if !inst.IsRunning() {
if !inst.IsRemote() && !inst.IsRunning() {
options := inst.GetOptions()
allowOnDemand := options != nil && options.OnDemandStart != nil && *options.OnDemandStart
if !allowOnDemand {
@@ -148,9 +136,6 @@ func (h *Handler) OpenAIProxy() http.HandlerFunc {
return
}
// Update last request time for the instance
inst.UpdateLastRequestTime()
// Recreate the request body from the bytes we read
r.Body = io.NopCloser(bytes.NewReader(bodyBytes))
r.ContentLength = int64(len(bodyBytes))
@@ -158,66 +143,3 @@ func (h *Handler) OpenAIProxy() http.HandlerFunc {
proxy.ServeHTTP(w, r)
}
}
// RemoteOpenAIProxy proxies OpenAI-compatible requests to a remote instance
func (h *Handler) RemoteOpenAIProxy(w http.ResponseWriter, r *http.Request, modelName string, inst *instance.Instance) {
// Get the node name from instance options
options := inst.GetOptions()
if options == nil {
http.Error(w, "Instance has no options configured", http.StatusInternalServerError)
return
}
// Get the first node from the set
var nodeName string
for node := range options.Nodes {
nodeName = node
break
}
if nodeName == "" {
http.Error(w, "Instance has no node configured", http.StatusInternalServerError)
return
}
// Check if we have a cached proxy for this node
h.remoteProxiesMu.RLock()
proxy, exists := h.remoteProxies[nodeName]
h.remoteProxiesMu.RUnlock()
if !exists {
// Find node configuration
nodeConfig, exists := h.cfg.Nodes[nodeName]
if !exists {
http.Error(w, fmt.Sprintf("Node %s not found", nodeName), http.StatusInternalServerError)
return
}
// Create reverse proxy to remote node
targetURL, err := url.Parse(nodeConfig.Address)
if err != nil {
http.Error(w, "Failed to parse node address: "+err.Error(), http.StatusInternalServerError)
return
}
proxy = httputil.NewSingleHostReverseProxy(targetURL)
// Modify request before forwarding
originalDirector := proxy.Director
apiKey := nodeConfig.APIKey // Capture for closure
proxy.Director = func(req *http.Request) {
originalDirector(req)
// Add API key if configured
if apiKey != "" {
req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", apiKey))
}
}
// Cache the proxy
h.remoteProxiesMu.Lock()
h.remoteProxies[nodeName] = proxy
h.remoteProxiesMu.Unlock()
}
// Forward the request using the cached proxy
proxy.ServeHTTP(w, r)
}