computing

package
v0.3.0 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Jul 11, 2026 License: MIT Imports: 42 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

View Source
var (
	ErrGlobalConcurrencyLimit = errors.New("global concurrency limit reached")
	ErrModelConcurrencyLimit  = errors.New("model concurrency limit reached")
)

Concurrency limit errors

View Source
var (
	ErrModelNotFound = &ModelError{Message: "model not found"}
)

Custom errors

Functions

func CheckMachineIdentity added in v0.2.0

func CheckMachineIdentity(cpRepoPath string) error

CheckMachineIdentity verifies that the private_key belongs to this machine. If the key was generated on a different machine (copied config), it prompts the user to regenerate a new node-id. In non-interactive environments, it returns an error.

func GenerateNodeID

func GenerateNodeID(cpRepoPath string) (string, string, string)

func GetNodeId

func GetNodeId(cpRepoPath string) string

Types

type AckPayload

type AckPayload struct {
	RequestID string `json:"request_id"`
	Success   bool   `json:"success"`
	Message   string `json:"message,omitempty"`
}

AckPayload for acknowledgments

type BenchmarkPayload added in v0.2.0

type BenchmarkPayload struct {
	BenchmarkID string            `json:"benchmark_id"`
	TestType    string            `json:"test_type"` // "math", "latency", "code", "reasoning"
	ModelID     string            `json:"model_id"`
	Prompts     []BenchmarkPrompt `json:"prompts"`
	TimeoutMs   int64             `json:"timeout_ms"`
}

BenchmarkPayload is sent from Swan Inference to run benchmark tests

type BenchmarkPrompt added in v0.2.0

type BenchmarkPrompt struct {
	ID             string `json:"id"`
	Prompt         string `json:"prompt"`
	ExpectedAnswer string `json:"expected_answer,omitempty"`
}

BenchmarkPrompt is a single prompt in a benchmark test

type BenchmarkResponsePayload added in v0.2.0

type BenchmarkResponsePayload struct {
	RequestID    string            `json:"request_id"`
	BenchmarkID  string            `json:"benchmark_id"`
	Results      []BenchmarkResult `json:"results"`
	TotalLatency int64             `json:"total_latency_ms"`
	Error        string            `json:"error,omitempty"`
}

BenchmarkResponsePayload is returned after processing a benchmark

type BenchmarkResult added in v0.2.0

type BenchmarkResult struct {
	PromptID  string `json:"prompt_id"`
	Answer    string `json:"answer"`
	Correct   bool   `json:"correct"`
	LatencyMs int64  `json:"latency_ms"`
	TokensIn  int64  `json:"tokens_in,omitempty"`
	TokensOut int64  `json:"tokens_out,omitempty"`
	Error     string `json:"error,omitempty"`
}

BenchmarkResult is the result of a single benchmark prompt

type ConcurrencyConfig

type ConcurrencyConfig struct {
	GlobalMaxConcurrent int           // Maximum concurrent requests globally
	DefaultModelMax     int           // Default max concurrent per model
	AcquireTimeout      time.Duration // Timeout for acquiring a slot
	EnableGPUAwareness  bool          // Adjust limits based on GPU memory
	GPUMemoryBufferMB   int           // Buffer to keep free in GPU memory
}

ConcurrencyConfig configures the concurrency limiter

func DefaultConcurrencyConfig

func DefaultConcurrencyConfig() ConcurrencyConfig

DefaultConcurrencyConfig returns sensible defaults

type ConcurrencyLimiter

type ConcurrencyLimiter struct {
	// contains filtered or unexported fields
}

ConcurrencyLimiter manages concurrent request limits

func NewConcurrencyLimiter

func NewConcurrencyLimiter(config ConcurrencyConfig, gpuCollector *GPUMetricsCollector) *ConcurrencyLimiter

NewConcurrencyLimiter creates a new concurrency limiter

func (*ConcurrencyLimiter) Acquire

func (cl *ConcurrencyLimiter) Acquire(ctx context.Context, modelID string) (*ConcurrencyToken, error)

Acquire acquires slots for a request (both global and model-specific)

func (*ConcurrencyLimiter) GetMetrics

func (cl *ConcurrencyLimiter) GetMetrics() ConcurrencyMetrics

GetMetrics returns concurrency metrics

func (*ConcurrencyLimiter) SetGlobalMax

func (cl *ConcurrencyLimiter) SetGlobalMax(max int)

SetGlobalMax updates the global maximum concurrent requests

func (*ConcurrencyLimiter) SetModelMax

func (cl *ConcurrencyLimiter) SetModelMax(modelID string, max int)

SetModelMax updates the maximum concurrent requests for a model

func (*ConcurrencyLimiter) Start

func (cl *ConcurrencyLimiter) Start()

Start begins the concurrency limiter

func (*ConcurrencyLimiter) Stop

func (cl *ConcurrencyLimiter) Stop()

Stop stops the concurrency limiter

type ConcurrencyMetrics

type ConcurrencyMetrics struct {
	GlobalActive   int64            `json:"global_active"`
	GlobalMax      int              `json:"global_max"`
	TotalAcquired  int64            `json:"total_acquired"`
	TotalReleased  int64            `json:"total_released"`
	TotalRejected  int64            `json:"total_rejected"`
	TotalTimeouts  int64            `json:"total_timeouts"`
	PerModelActive map[string]int64 `json:"per_model_active"`
	PerModelMax    map[string]int   `json:"per_model_max"`
	AvgHoldTimeMs  float64          `json:"avg_hold_time_ms"`
}

ConcurrencyMetrics tracks concurrency statistics

type ConcurrencyToken

type ConcurrencyToken struct {
	// contains filtered or unexported fields
}

ConcurrencyToken represents an acquired concurrency slot

func (*ConcurrencyToken) Release

func (ct *ConcurrencyToken) Release()

Release releases the concurrency slots

type DeterministicChallengeData

type DeterministicChallengeData struct {
	Prompt    string `json:"prompt"`
	Seed      int    `json:"seed"`
	MaxTokens int    `json:"max_tokens"`
}

DeterministicChallengeData represents a deterministic inference challenge from the server

type DeterministicResponseData

type DeterministicResponseData struct {
	Tokens []string `json:"tokens"`
	Text   string   `json:"text"`
}

DeterministicResponseData is the response sent back for a deterministic challenge

type ErrorPayload

type ErrorPayload struct {
	RequestID string `json:"request_id,omitempty"`
	Code      int    `json:"code"`
	Message   string `json:"message"`
}

ErrorPayload for error responses

type FingerprintChallengeData

type FingerprintChallengeData struct {
	Files []FingerprintChallengeFile `json:"files"`
}

FingerprintChallengeData represents the fingerprint challenge from the server

type FingerprintChallengeFile

type FingerprintChallengeFile struct {
	Filename     string `json:"filename"`
	ExpectedHash string `json:"expected_hash"`
}

FingerprintChallengeFile is a single file in a fingerprint challenge

type FingerprintResponseData

type FingerprintResponseData struct {
	Files []FingerprintResponseFile `json:"files"`
}

FingerprintResponseData is the response sent back for a fingerprint challenge

type FingerprintResponseFile

type FingerprintResponseFile struct {
	Filename string `json:"filename"`
	Hash     string `json:"hash"`
	Status   string `json:"status"` // "pass", "fail", "missing"
}

FingerprintResponseFile is a single file result in a fingerprint response

type GPUMetrics

type GPUMetrics struct {
	Index            int     `json:"index"`
	Name             string  `json:"name"`
	UUID             string  `json:"uuid,omitempty"`
	UtilizationPct   float64 `json:"utilization_percent"`
	MemoryUsedMB     float64 `json:"memory_used_mb"`
	MemoryTotalMB    float64 `json:"memory_total_mb"`
	MemoryUsagePct   float64 `json:"memory_usage_percent"`
	TemperatureC     float64 `json:"temperature_c"`
	PowerDrawW       float64 `json:"power_draw_w"`
	PowerLimitW      float64 `json:"power_limit_w"`
	FanSpeedPct      float64 `json:"fan_speed_percent,omitempty"`
	ComputeProcesses int     `json:"compute_processes"`
}

GPUMetrics tracks metrics for a single GPU

type GPUMetricsCollector

type GPUMetricsCollector struct{}

GPUMetricsCollector collects real-time GPU metrics using nvidia-smi

func NewGPUMetricsCollector

func NewGPUMetricsCollector() *GPUMetricsCollector

NewGPUMetricsCollector creates a new GPU metrics collector

func (*GPUMetricsCollector) CollectGPUMetrics

func (c *GPUMetricsCollector) CollectGPUMetrics() []GPUMetrics

CollectGPUMetrics collects real-time metrics from all available GPUs

func (*GPUMetricsCollector) GetAggregatedGPUMetrics

func (c *GPUMetricsCollector) GetAggregatedGPUMetrics() (avgUtilization, avgMemoryUsage float64)

GetAggregatedGPUMetrics returns aggregated metrics across all GPUs

type HardwareInfo

type HardwareInfo struct {
	GPUType           string `json:"gpu_type"`
	GPUModel          string `json:"gpu_model"`
	VRAMGB            int    `json:"vram_gb"`
	GPUCount          int    `json:"gpu_count"`
	ComputeCapability string `json:"compute_capability"`
	DriverVersion     string `json:"driver_version"`
	CUDAVersion       string `json:"cuda_version"`
	ServingEngine     string `json:"serving_engine,omitempty"` // "vllm", "sglang", "llamacpp", "ollama", "tgi", "unknown"
}

HardwareInfo contains GPU hardware specifications

func DetectGPUHardware

func DetectGPUHardware() *HardwareInfo

DetectGPUHardware detects GPU hardware information

type HealthCheckConfig

type HealthCheckConfig struct {
	Interval           time.Duration // How often to check health
	Timeout            time.Duration // Timeout for each health check
	UnhealthyThreshold int           // Consecutive failures before marking unhealthy
	HealthyThreshold   int           // Consecutive successes to recover from unhealthy
	CircuitOpenTime    time.Duration // How long to keep circuit open before retrying
}

HealthCheckConfig configures the health checker behavior

func DefaultHealthCheckConfig

func DefaultHealthCheckConfig() HealthCheckConfig

DefaultHealthCheckConfig returns default health check configuration

type HeartbeatPayload

type HeartbeatPayload struct {
	NodeID      string             `json:"node_id"`               // Local node ID (not the DB provider ID)
	ProviderID  string             `json:"provider_id,omitempty"` // Deprecated: use NodeID
	Timestamp   int64              `json:"timestamp"`
	Metrics     map[string]float64 `json:"metrics,omitempty"`
	Models      []string           `json:"models,omitempty"`       // Current model list (allows dynamic model updates without reconnect)
	ModelHealth map[string]string  `json:"model_health,omitempty"` // modelID -> health status (backup for health updates)
	Hardware    *HardwareInfo      `json:"hardware,omitempty"`     // GPU hardware info (periodically updated)
}

HeartbeatPayload for liveness checks

type HistoricalDataPoint

type HistoricalDataPoint struct {
	Timestamp         time.Time `json:"timestamp"`
	TotalRequests     int64     `json:"total_requests"`
	SuccessRate       float64   `json:"success_rate"`
	AvgLatencyMs      float64   `json:"avg_latency_ms"`
	P99LatencyMs      float64   `json:"p99_latency_ms"`
	TokensPerSecond   float64   `json:"tokens_per_second"`
	RequestsPerMinute float64   `json:"requests_per_minute"`
}

HistoricalDataPoint represents an aggregated data point for API responses

type HttpClient

type HttpClient struct {
	// contains filtered or unexported fields
}

func NewHttpClient

func NewHttpClient(host string, header http.Header) *HttpClient

func (*HttpClient) PostJSON

func (c *HttpClient) PostJSON(api string, data any, dest any) error

func (*HttpClient) Request

func (c *HttpClient) Request(method string, api string, body io.Reader, dest any, contentType ...string) (err error)

type InferenceClient

type InferenceClient struct {
	// contains filtered or unexported fields
}

InferenceClient manages WebSocket connection to Swan Inference service

func NewInferenceClient

func NewInferenceClient(nodeID, workerAddr, ownerAddr string) *InferenceClient

NewInferenceClient creates a new Inference client

func (*InferenceClient) GetMetrics

func (c *InferenceClient) GetMetrics() InferenceMetricsData

GetMetrics returns a snapshot of the current metrics

func (*InferenceClient) GetMetricsPrometheus

func (c *InferenceClient) GetMetricsPrometheus() string

GetMetricsPrometheus returns metrics in Prometheus text format

func (*InferenceClient) IsConnected

func (c *InferenceClient) IsConnected() bool

IsConnected returns whether the client is connected, registered, and healthy. A connection is considered unhealthy if 3+ consecutive heartbeats went unacknowledged.

func (*InferenceClient) SendModelHealthUpdate

func (c *InferenceClient) SendModelHealthUpdate(modelHealth map[string]string)

SendModelHealthUpdate sends a model health update to Swan Inference This is called when model health status changes (healthy/degraded/unhealthy)

func (*InferenceClient) SetInferenceHandler

func (c *InferenceClient) SetInferenceHandler(handler InferenceHandler)

SetInferenceHandler sets the handler for non-streaming inference requests

func (*InferenceClient) SetModelHealthProvider

func (c *InferenceClient) SetModelHealthProvider(provider func() map[string]string)

SetModelHealthProvider sets the function that provides current model health for heartbeats

func (*InferenceClient) SetModelMappingsProvider

func (c *InferenceClient) SetModelMappingsProvider(provider func() map[string]ModelMapping)

SetModelMappingsProvider sets the function that returns model mappings for format/quantization

func (*InferenceClient) SetStreamingInferenceHandler

func (c *InferenceClient) SetStreamingInferenceHandler(handler StreamingInferenceHandler)

SetStreamingInferenceHandler sets the handler for streaming inference requests

func (*InferenceClient) SetWarmupHandler

func (c *InferenceClient) SetWarmupHandler(handler WarmupHandler)

SetWarmupHandler sets the handler for model warmup requests

func (*InferenceClient) Start

func (c *InferenceClient) Start() error

Start connects to Swan Inference and starts the client

func (*InferenceClient) Stop

func (c *InferenceClient) Stop()

Stop gracefully shuts down the client

type InferenceHandler

type InferenceHandler func(payload InferencePayload) (*InferenceResponse, error)

InferenceHandler handles non-streaming inference requests from Inference service

type InferenceMetrics

type InferenceMetrics struct {
	InferenceMetricsData
	// contains filtered or unexported fields
}

InferenceMetrics tracks metrics for the inference service

func NewInferenceMetrics

func NewInferenceMetrics() *InferenceMetrics

NewInferenceMetrics creates a new InferenceMetrics instance

func (*InferenceMetrics) GetPrometheusMetrics

func (m *InferenceMetrics) GetPrometheusMetrics() string

GetPrometheusMetrics returns metrics in Prometheus text format

func (*InferenceMetrics) GetRequestHistory

func (m *InferenceMetrics) GetRequestHistory(limit int, modelFilter string) []RequestMetric

GetRequestHistory returns recent requests, optionally filtered by model

func (*InferenceMetrics) GetSnapshot

func (m *InferenceMetrics) GetSnapshot() InferenceMetricsData

GetSnapshot returns a copy of the current metrics

func (*InferenceMetrics) RecordConnectionState

func (m *InferenceMetrics) RecordConnectionState(state string)

RecordConnectionState updates the connection state

func (*InferenceMetrics) RecordReconnect

func (m *InferenceMetrics) RecordReconnect()

RecordReconnect increments the reconnect counter

func (*InferenceMetrics) RecordRequest

func (m *InferenceMetrics) RecordRequest(req RequestMetric)

RecordRequest adds a request to the history circular buffer

func (*InferenceMetrics) RecordRequestEnd

func (m *InferenceMetrics) RecordRequestEnd(req RequestMetric)

RecordRequestEnd records the completion of a request and appends it to the request history buffer (served by the /inference/requests endpoint)

func (*InferenceMetrics) RecordRequestStart

func (m *InferenceMetrics) RecordRequestStart(model string, streaming bool)

RecordRequestStart records the start of a request

func (*InferenceMetrics) UpdateGPUMetrics

func (m *InferenceMetrics) UpdateGPUMetrics(gpuMetrics []GPUMetrics)

UpdateGPUMetrics updates the GPU metrics

type InferenceMetricsData added in v0.3.0

type InferenceMetricsData struct {
	// Connection metrics
	ConnectionState    string    `json:"connection_state"`
	LastConnectedAt    time.Time `json:"last_connected_at,omitempty"`
	LastDisconnectedAt time.Time `json:"last_disconnected_at,omitempty"`
	ReconnectCount     int64     `json:"reconnect_count"`

	// Request metrics (aggregated)
	TotalRequests     int64   `json:"total_requests"`
	SuccessfulReqs    int64   `json:"successful_requests"`
	FailedReqs        int64   `json:"failed_requests"`
	StreamingReqs     int64   `json:"streaming_requests"`
	AvgLatencyMs      float64 `json:"avg_latency_ms"`
	P50LatencyMs      float64 `json:"p50_latency_ms"`
	P95LatencyMs      float64 `json:"p95_latency_ms"`
	P99LatencyMs      float64 `json:"p99_latency_ms"`
	TotalTokensIn     int64   `json:"total_tokens_in"`
	TotalTokensOut    int64   `json:"total_tokens_out"`
	TokensPerSecond   float64 `json:"tokens_per_second"`
	ActiveRequests    int64   `json:"active_requests"`
	RequestsPerMinute float64 `json:"requests_per_minute"`

	// Per-model metrics
	ModelMetrics map[string]*ModelMetrics `json:"model_metrics"`

	// GPU metrics
	GPUMetrics []GPUMetrics `json:"gpu_metrics"`

	// System metrics
	CPUUsagePercent    float64 `json:"cpu_usage_percent"`
	MemoryUsagePercent float64 `json:"memory_usage_percent"`
	MemoryUsedGB       float64 `json:"memory_used_gb"`
	MemoryTotalGB      float64 `json:"memory_total_gb"`
}

InferenceMetricsData holds the exported metric values. It contains no locks, so it can be copied freely (e.g. returned as a snapshot).

type InferencePayload

type InferencePayload struct {
	EndpointID string          `json:"endpoint_id"`
	ModelID    string          `json:"model_id"`
	Request    json.RawMessage `json:"request"`
	Stream     bool            `json:"stream"` // Whether to stream the response
}

InferencePayload is sent to provider for inference request

type InferenceResponse

type InferenceResponse struct {
	RequestID  string          `json:"request_id"`
	Response   json.RawMessage `json:"response"`
	Error      string          `json:"error,omitempty"`
	StatusCode int             `json:"status_code,omitempty"` // HTTP status code for Swan Inference to map to proper responses
	Latency    int64           `json:"latency_ms"`
}

InferenceResponse is returned by provider

type InferenceService

type InferenceService struct {
	// contains filtered or unexported fields
}

InferenceService manages the Inference client and inference handling

func NewInferenceService

func NewInferenceService(nodeID, cpPath string) *InferenceService

NewInferenceService creates a new Inference service

func (*InferenceService) DisableModel

func (s *InferenceService) DisableModel(modelID string) error

DisableModel disables a model from serving requests

func (*InferenceService) EnableModel

func (s *InferenceService) EnableModel(modelID string) error

EnableModel enables a model for serving requests

func (*InferenceService) ForceHealthCheck

func (s *InferenceService) ForceHealthCheck(modelID string)

ForceHealthCheck triggers an immediate health check for a model

func (*InferenceService) GetActiveModels

func (s *InferenceService) GetActiveModels() []string

GetActiveModels returns the list of active model deployments

func (*InferenceService) GetAllModelHealth

func (s *InferenceService) GetAllModelHealth() map[string]*ModelStatus

GetAllModelHealth returns health status of all models

func (*InferenceService) GetAllModels

func (s *InferenceService) GetAllModels() []*RegisteredModel

GetAllModels returns all registered models with their status

func (*InferenceService) GetConcurrencyMetrics

func (s *InferenceService) GetConcurrencyMetrics() *ConcurrencyMetrics

GetConcurrencyMetrics returns concurrency limiter metrics

func (*InferenceService) GetMetrics

func (s *InferenceService) GetMetrics() *InferenceMetricsData

GetMetrics returns a snapshot of the current inference metrics

func (*InferenceService) GetMetricsHistory

func (s *InferenceService) GetMetricsHistory(duration, resolution time.Duration) ([]HistoricalDataPoint, error)

GetMetricsHistory returns historical metrics for the specified duration and resolution

func (*InferenceService) GetMetricsPrometheus

func (s *InferenceService) GetMetricsPrometheus() string

GetMetricsPrometheus returns metrics in Prometheus text format

func (*InferenceService) GetModelDetailedMetrics

func (s *InferenceService) GetModelDetailedMetrics(modelID string) map[string]interface{}

GetModelDetailedMetrics returns detailed metrics for a specific model including recent requests

func (*InferenceService) GetModelHealth

func (s *InferenceService) GetModelHealth(modelID string) (*ModelStatus, bool)

GetModelHealth returns the health status of a specific model

func (*InferenceService) GetModelStatus

func (s *InferenceService) GetModelStatus(modelID string) (*RegisteredModel, bool)

GetModelStatus returns the status of a specific model

func (*InferenceService) GetModelsSummary

func (s *InferenceService) GetModelsSummary() map[string]interface{}

GetModelsSummary returns a summary of model statuses

func (*InferenceService) GetRateLimiterMetrics

func (s *InferenceService) GetRateLimiterMetrics() *RateLimiterMetrics

GetRateLimiterMetrics returns rate limiter metrics

func (*InferenceService) GetRequestHistory

func (s *InferenceService) GetRequestHistory(limit int, modelFilter string) []RequestMetric

GetRequestHistory returns recent request history, optionally filtered by model

func (*InferenceService) GetRequestManagementStatus

func (s *InferenceService) GetRequestManagementStatus() map[string]interface{}

GetRequestManagementStatus returns combined status of all request management components

func (*InferenceService) GetRetryMetrics

func (s *InferenceService) GetRetryMetrics() *RetryMetrics

GetRetryMetrics returns retry policy metrics

func (*InferenceService) IsConnected

func (s *InferenceService) IsConnected() bool

IsConnected returns whether the Inference client is connected

func (*InferenceService) ReloadModels

func (s *InferenceService) ReloadModels() error

ReloadModels manually triggers a reload of the models configuration

func (*InferenceService) SetGlobalConcurrencyLimit

func (s *InferenceService) SetGlobalConcurrencyLimit(max int)

SetGlobalConcurrencyLimit updates the global concurrency limit

func (*InferenceService) SetGlobalRateLimit

func (s *InferenceService) SetGlobalRateLimit(tokensPerSecond float64)

SetGlobalRateLimit updates the global rate limit

func (*InferenceService) SetModelConcurrencyLimit

func (s *InferenceService) SetModelConcurrencyLimit(modelID string, max int)

SetModelConcurrencyLimit sets concurrency limit for a specific model

func (*InferenceService) SetModelRateLimit

func (s *InferenceService) SetModelRateLimit(modelID string, tokensPerSecond float64, burstSize int)

SetModelRateLimit sets rate limit for a specific model

func (*InferenceService) Start

func (s *InferenceService) Start() error

Start initializes and starts the Inference client

func (*InferenceService) Stop

func (s *InferenceService) Stop()

Stop gracefully shuts down the Inference service

type Message

type Message struct {
	Type      MessageType     `json:"type"`
	RequestID string          `json:"request_id,omitempty"`
	Payload   json.RawMessage `json:"payload"`
}

Message is the base WebSocket message structure

type MessageType

type MessageType string

Inference WebSocket Protocol Types

const (
	MsgTypeRegister          MessageType = "register"
	MsgTypeInference         MessageType = "inference"
	MsgTypeVerify            MessageType = "verify"
	MsgTypeHeartbeat         MessageType = "heartbeat"
	MsgTypeAck               MessageType = "ack"
	MsgTypeError             MessageType = "error"
	MsgTypeStreamChunk       MessageType = "stream_chunk"        // Streaming chunk to Swan Inference
	MsgTypeStreamEnd         MessageType = "stream_end"          // End of stream marker
	MsgTypeWarmup            MessageType = "warmup"              // Model warmup request
	MsgTypeModelHealthUpdate MessageType = "model_health_update" // Model health status update
	MsgTypeBenchmark         MessageType = "benchmark"           // Benchmark test request from server
	MsgTypeBenchmarkResponse MessageType = "benchmark_response"  // Benchmark test results to server
)

type MetricsHistory

type MetricsHistory struct {
	// contains filtered or unexported fields
}

MetricsHistory manages historical metrics storage and retrieval

func NewMetricsHistory

func NewMetricsHistory() *MetricsHistory

NewMetricsHistory creates a new MetricsHistory instance

func (*MetricsHistory) GetHistory

func (h *MetricsHistory) GetHistory(duration time.Duration, resolution time.Duration) ([]HistoricalDataPoint, error)

GetHistory retrieves historical metrics for the given duration with the specified resolution

func (*MetricsHistory) Start

func (h *MetricsHistory) Start(metricsProvider func() *InferenceMetricsData) error

Start begins the metrics recording goroutine

func (*MetricsHistory) Stop

func (h *MetricsHistory) Stop()

Stop stops the metrics recording goroutine

type MetricsHistoryEntity

type MetricsHistoryEntity struct {
	ID                uint      `gorm:"primaryKey;autoIncrement"`
	Timestamp         time.Time `gorm:"index;not null"`
	TotalRequests     int64     `json:"total_requests"`
	SuccessfulReqs    int64     `json:"successful_requests"`
	FailedReqs        int64     `json:"failed_requests"`
	SuccessRate       float64   `json:"success_rate"`
	AvgLatencyMs      float64   `json:"avg_latency_ms"`
	P50LatencyMs      float64   `json:"p50_latency_ms"`
	P95LatencyMs      float64   `json:"p95_latency_ms"`
	P99LatencyMs      float64   `json:"p99_latency_ms"`
	TokensPerSecond   float64   `json:"tokens_per_second"`
	RequestsPerMinute float64   `json:"requests_per_minute"`
	ActiveRequests    int64     `json:"active_requests"`
	TotalTokensIn     int64     `json:"total_tokens_in"`
	TotalTokensOut    int64     `json:"total_tokens_out"`
}

MetricsHistoryEntity represents a historical metrics data point in the database

func (MetricsHistoryEntity) TableName

func (MetricsHistoryEntity) TableName() string

type ModelError

type ModelError struct {
	Message string
}

func (*ModelError) Error

func (e *ModelError) Error() string

type ModelHealth

type ModelHealth int

ModelHealth represents the health state of a model endpoint

const (
	ModelHealthUnknown ModelHealth = iota
	ModelHealthHealthy
	ModelHealthDegraded
	ModelHealthUnhealthy
)

func (ModelHealth) String

func (h ModelHealth) String() string

type ModelHealthChecker

type ModelHealthChecker struct {
	// contains filtered or unexported fields
}

ModelHealthChecker performs periodic health checks on model endpoints

func NewModelHealthChecker

func NewModelHealthChecker(config HealthCheckConfig) *ModelHealthChecker

NewModelHealthChecker creates a new health checker

func (*ModelHealthChecker) ForceCheck

func (h *ModelHealthChecker) ForceCheck(modelID string)

ForceCheck triggers an immediate health check for a model

func (*ModelHealthChecker) GetAllStatuses

func (h *ModelHealthChecker) GetAllStatuses() map[string]*ModelStatus

GetAllStatuses returns health status of all models

func (*ModelHealthChecker) GetModelStatus

func (h *ModelHealthChecker) GetModelStatus(modelID string) (*ModelStatus, bool)

GetModelStatus returns the health status of a specific model

func (*ModelHealthChecker) IsModelHealthy

func (h *ModelHealthChecker) IsModelHealthy(modelID string) bool

IsModelHealthy returns whether a specific model is healthy enough to serve requests

func (*ModelHealthChecker) RegisterModel

func (h *ModelHealthChecker) RegisterModel(modelID, endpoint, apiKey string)

RegisterModel adds a model to health checking

func (*ModelHealthChecker) SetStatusChangeCallback

func (h *ModelHealthChecker) SetStatusChangeCallback(cb func(modelID string, oldHealth, newHealth ModelHealth))

SetStatusChangeCallback sets a callback for health status changes

func (*ModelHealthChecker) Start

func (h *ModelHealthChecker) Start()

Start begins periodic health checking

func (*ModelHealthChecker) Stop

func (h *ModelHealthChecker) Stop()

Stop stops the health checker

func (*ModelHealthChecker) UnregisterModel

func (h *ModelHealthChecker) UnregisterModel(modelID string)

UnregisterModel removes a model from health checking

type ModelHealthUpdatePayload

type ModelHealthUpdatePayload struct {
	NodeID      string            `json:"node_id"`               // Local node ID (not the DB provider ID)
	ProviderID  string            `json:"provider_id,omitempty"` // Deprecated: use NodeID
	ModelHealth map[string]string `json:"model_health"`          // modelID -> health status ("healthy", "degraded", "unhealthy")
	Timestamp   int64             `json:"timestamp"`
}

ModelHealthUpdatePayload is sent to Swan Inference when model health changes

type ModelInfo

type ModelInfo struct {
	ModelID      string `json:"model_id"`
	WeightHash   string `json:"weight_hash,omitempty"`  // Composite SHA256 of all weight files
	HashAlgo     string `json:"hash_algo,omitempty"`    // Hash algorithm, e.g. "sha256"
	Format       string `json:"format,omitempty"`       // Weight format: "fp16", "fp8", "awq", "gptq", "gguf"
	Quantization string `json:"quantization,omitempty"` // Quantization detail: "q4_k_m", "q8_0", "w4a16", etc.
}

ModelInfo contains model identification and verification hash

type ModelMapping

type ModelMapping struct {
	Container    string `json:"container"`
	Endpoint     string `json:"endpoint"`
	GPUMemory    int    `json:"gpu_memory"`
	Category     string `json:"category"`
	LocalModel   string `json:"local_model"`            // Actual model name for local inference server (e.g., Ollama model name)
	Format       string `json:"format,omitempty"`       // Weight format: "fp16", "fp8", "awq", "gptq", "gguf"
	Quantization string `json:"quantization,omitempty"` // Quantization detail: "q4_k_m", "q8_0", "w4a16", etc.
	APIKey       string `json:"api_key,omitempty"`      // API key for authenticated model endpoints (e.g., vLLM --api-key)
}

ModelMapping represents a model-to-endpoint mapping from models.json

type ModelMetrics

type ModelMetrics struct {
	ModelName       string  `json:"model_name"`
	TotalRequests   int64   `json:"total_requests"`
	SuccessfulReqs  int64   `json:"successful_requests"`
	FailedReqs      int64   `json:"failed_requests"`
	AvgLatencyMs    float64 `json:"avg_latency_ms"`
	TotalTokensIn   int64   `json:"total_tokens_in"`
	TotalTokensOut  int64   `json:"total_tokens_out"`
	TokensPerSecond float64 `json:"tokens_per_second"`
	ActiveRequests  int64   `json:"active_requests"`
	// contains filtered or unexported fields
}

ModelMetrics tracks metrics for a specific model

type ModelRegistry

type ModelRegistry struct {
	// contains filtered or unexported fields
}

ModelRegistry manages the lifecycle of model configurations

func NewModelRegistry

func NewModelRegistry(configPath string, healthChecker *ModelHealthChecker) *ModelRegistry

NewModelRegistry creates a new model registry

func (*ModelRegistry) DisableModel

func (r *ModelRegistry) DisableModel(modelID string) error

DisableModel disables a model from serving

func (*ModelRegistry) EnableModel

func (r *ModelRegistry) EnableModel(modelID string) error

EnableModel enables a model for serving

func (*ModelRegistry) GetAllModelHealthMap

func (r *ModelRegistry) GetAllModelHealthMap() map[string]string

GetAllModelHealthMap returns a map of all model health statuses Returns modelID -> health status string ("healthy", "degraded", "unhealthy", "unknown")

func (*ModelRegistry) GetAllModels

func (r *ModelRegistry) GetAllModels() []*RegisteredModel

GetAllModels returns all registered models

func (*ModelRegistry) GetLocalModelName

func (r *ModelRegistry) GetLocalModelName(modelID string) string

GetLocalModelName returns the local model name for a model (e.g., Ollama model name) Returns empty string if not configured (use the model ID directly)

func (*ModelRegistry) GetModel

func (r *ModelRegistry) GetModel(modelID string) (*RegisteredModel, bool)

GetModel returns a registered model by ID

func (*ModelRegistry) GetModelAPIKey

func (r *ModelRegistry) GetModelAPIKey(modelID string) string

GetModelAPIKey returns the API key for a model endpoint Returns empty string if not configured

func (*ModelRegistry) GetModelEndpoint

func (r *ModelRegistry) GetModelEndpoint(modelID string) (string, bool)

GetModelEndpoint returns the endpoint for a model if it's ready

func (*ModelRegistry) GetReadyModelIDs

func (r *ModelRegistry) GetReadyModelIDs() []string

GetReadyModelIDs returns IDs of models ready to serve requests

func (*ModelRegistry) GetReadyModels

func (r *ModelRegistry) GetReadyModels() []*RegisteredModel

GetReadyModels returns models that are ready to serve requests

func (*ModelRegistry) GetStatusSummary

func (r *ModelRegistry) GetStatusSummary() map[string]interface{}

GetStatusSummary returns a summary of model statuses

func (*ModelRegistry) ReloadConfig

func (r *ModelRegistry) ReloadConfig() error

ReloadConfig manually triggers a configuration reload

func (*ModelRegistry) SetCallbacks

func (r *ModelRegistry) SetCallbacks(
	onAdded func(model *RegisteredModel),
	onRemoved func(modelID string),
	onUpdated func(model *RegisteredModel),
)

SetCallbacks sets callbacks for model lifecycle events

func (*ModelRegistry) SetHealthUpdateCallback

func (r *ModelRegistry) SetHealthUpdateCallback(callback func(modelHealth map[string]string))

SetHealthUpdateCallback sets the callback for model health updates The callback receives a map of modelID -> health status ("healthy", "degraded", "unhealthy")

func (*ModelRegistry) Start

func (r *ModelRegistry) Start() error

Start loads initial configuration and begins watching for changes

func (*ModelRegistry) Stop

func (r *ModelRegistry) Stop()

Stop stops the registry and file watcher

type ModelServerError

type ModelServerError struct {
	StatusCode int    // HTTP status code from the model server
	Body       []byte // Raw response body
	Message    string // Parsed error message (from OpenAI error format or raw body)
}

ModelServerError represents a non-2xx HTTP response from the model server. It preserves the original status code and body so callers can propagate meaningful error codes (e.g. 400, 404, 429, 503) to Swan Inference.

func (*ModelServerError) Error

func (e *ModelServerError) Error() string

type ModelState

type ModelState int

ModelState represents the current state of a model

const (
	ModelStateUnknown ModelState = iota
	ModelStateLoading
	ModelStateReady
	ModelStateUnhealthy
	ModelStateDisabled
)

func (ModelState) String

func (s ModelState) String() string

type ModelStatus

type ModelStatus struct {
	ModelID          string      `json:"model_id"`
	Endpoint         string      `json:"endpoint"`
	Health           ModelHealth `json:"health"`
	HealthString     string      `json:"health_string"`
	LastCheck        time.Time   `json:"last_check"`
	LastSuccess      time.Time   `json:"last_success"`
	LastError        string      `json:"last_error,omitempty"`
	LatencyMs        float64     `json:"latency_ms"`
	AvgLatencyMs     float64     `json:"avg_latency_ms"`
	ConsecutiveFails int         `json:"consecutive_fails"`
	TotalChecks      int64       `json:"total_checks"`
	TotalSuccesses   int64       `json:"total_successes"`
	TotalFailures    int64       `json:"total_failures"`
	CircuitOpen      bool        `json:"circuit_open"`
}

ModelStatus tracks the health status of a single model

type ProviderStatusResponse

type ProviderStatusResponse struct {
	ProviderID      string   `json:"provider_id"`
	Name            string   `json:"name"`
	Status          string   `json:"status"`
	CanConnect      bool     `json:"can_connect"`
	APIKeyValid     bool     `json:"api_key_valid"`
	Message         string   `json:"message"`
	Warning         string   `json:"warning,omitempty"`
	NextSteps       []string `json:"next_steps,omitempty"`
	Step            int      `json:"step"`
	TotalSteps      int      `json:"total_steps"`
	StepLabel       string   `json:"step_label"`
	EarningsEnabled bool     `json:"earnings_enabled"`
}

ProviderStatusResponse represents the status check response from Swan Inference

type RateLimiter

type RateLimiter struct {
	// contains filtered or unexported fields
}

RateLimiter provides rate limiting with optional adaptive adjustment

func NewRateLimiter

func NewRateLimiter(config RateLimiterConfig, gpuCollector *GPUMetricsCollector) *RateLimiter

NewRateLimiter creates a new rate limiter

func (*RateLimiter) AllowModel

func (rl *RateLimiter) AllowModel(modelID string) bool

AllowModel checks if a request for a specific model is allowed

func (*RateLimiter) GetMetrics

func (rl *RateLimiter) GetMetrics() RateLimiterMetrics

GetMetrics returns rate limiter metrics

func (*RateLimiter) SetModelLimit

func (rl *RateLimiter) SetModelLimit(modelID string, tokensPerSecond float64, burstSize int)

SetModelLimit sets a rate limit for a specific model

func (*RateLimiter) Start

func (rl *RateLimiter) Start()

Start begins the rate limiter (adaptive adjustment if enabled)

func (*RateLimiter) Stop

func (rl *RateLimiter) Stop()

Stop stops the rate limiter

type RateLimiterConfig

type RateLimiterConfig struct {
	// Token bucket settings
	TokensPerSecond float64 // Rate of token replenishment
	BurstSize       int     // Maximum burst capacity

	// Adaptive rate limiting
	EnableAdaptive     bool    // Enable GPU-aware rate limiting
	GPUThresholdHigh   float64 // GPU utilization above which to reduce rate
	GPUThresholdLow    float64 // GPU utilization below which to increase rate
	AdaptiveMinRate    float64 // Minimum tokens per second when adapting
	AdaptiveMaxRate    float64 // Maximum tokens per second when adapting
	AdaptiveAdjustment float64 // Rate adjustment factor per interval
}

RateLimiterConfig configures the rate limiter

func DefaultRateLimiterConfig

func DefaultRateLimiterConfig() RateLimiterConfig

DefaultRateLimiterConfig returns sensible defaults

type RateLimiterMetrics

type RateLimiterMetrics struct {
	TotalAllowed    int64   `json:"total_allowed"`
	TotalThrottled  int64   `json:"total_throttled"`
	CurrentRate     float64 `json:"current_rate"`
	CurrentTokens   float64 `json:"current_tokens"`
	BurstSize       int     `json:"burst_size"`
	AdaptiveEnabled bool    `json:"adaptive_enabled"`
}

RateLimiterMetrics tracks rate limiter statistics

type RegisterPayload

type RegisterPayload struct {
	NodeID       string        `json:"node_id"`               // Local node ID (not the DB provider ID)
	ProviderID   string        `json:"provider_id,omitempty"` // Deprecated: use NodeID
	NodeName     string        `json:"node_name,omitempty"`   // Human-readable provider name from config
	WorkerAddr   string        `json:"worker_addr"`
	OwnerAddr    string        `json:"owner_addr"`
	Token        string        `json:"token,omitempty"` // API key for authentication (sk-prov-*)
	Signature    string        `json:"signature,omitempty"`
	Models       []string      `json:"models"`
	ModelHashes  []ModelInfo   `json:"model_hashes,omitempty"` // Per-model composite hashes for verification
	Capabilities []string      `json:"capabilities"`
	Hardware     *HardwareInfo `json:"hardware,omitempty"`
}

RegisterPayload is sent by provider on connection

type RegisteredModel

type RegisteredModel struct {
	ID           string      `json:"id"`
	Container    string      `json:"container"`
	Endpoint     string      `json:"endpoint"`
	GPUMemory    int         `json:"gpu_memory"`
	Category     string      `json:"category"`
	LocalModel   string      `json:"local_model,omitempty"`  // Actual model name for local inference server
	Format       string      `json:"format,omitempty"`       // Weight format: fp16, awq, gptq, gguf, etc.
	Quantization string      `json:"quantization,omitempty"` // Quantization detail: q4_k_m, q8_0, w4a16, etc.
	APIKey       string      `json:"api_key,omitempty"`      // API key for authenticated model endpoints
	State        ModelState  `json:"state"`
	StateString  string      `json:"state_string"`
	Health       ModelHealth `json:"health"`
	HealthString string      `json:"health_string"`
	LoadedAt     time.Time   `json:"loaded_at,omitempty"`
	UpdatedAt    time.Time   `json:"updated_at"`
	Enabled      bool        `json:"enabled"`
}

RegisteredModel represents a fully configured model in the registry

type RequestMetric

type RequestMetric struct {
	RequestID   string    `json:"request_id"`
	Model       string    `json:"model"`
	StartTime   time.Time `json:"start_time"`
	EndTime     time.Time `json:"end_time,omitempty"`
	LatencyMs   float64   `json:"latency_ms"`
	TokensIn    int       `json:"tokens_in"`
	TokensOut   int       `json:"tokens_out"`
	Streaming   bool      `json:"streaming"`
	Success     bool      `json:"success"`
	ErrorReason string    `json:"error_reason,omitempty"`
}

RequestMetric represents a single request's metrics

type ResultChecker

type ResultChecker interface {
	Check() error
}

type RetryConfig

type RetryConfig struct {
	MaxRetries         int           // Maximum number of retry attempts
	InitialDelay       time.Duration // Initial delay before first retry
	MaxDelay           time.Duration // Maximum delay between retries
	Multiplier         float64       // Delay multiplier for exponential backoff
	JitterFactor       float64       // Random jitter factor (0-1)
	RetryableErrors    []string      // Error substrings that are retryable
	NonRetryableErrors []string      // Error substrings that should not be retried
}

RetryConfig configures retry behavior

func DefaultRetryConfig

func DefaultRetryConfig() RetryConfig

DefaultRetryConfig returns sensible defaults

type RetryMetrics

type RetryMetrics struct {
	TotalAttempts        int64   `json:"total_attempts"`
	TotalRetries         int64   `json:"total_retries"`
	TotalSuccesses       int64   `json:"total_successes"`
	TotalFailures        int64   `json:"total_failures"`
	TotalNonRetryable    int64   `json:"total_non_retryable"`
	AvgRetriesPerRequest float64 `json:"avg_retries_per_request"`
	RetrySuccessRate     float64 `json:"retry_success_rate"`
}

RetryMetrics tracks retry statistics

type RetryPolicy

type RetryPolicy struct {
	// contains filtered or unexported fields
}

RetryPolicy implements retry logic with exponential backoff and jitter

func NewRetryPolicy

func NewRetryPolicy(config RetryConfig) *RetryPolicy

NewRetryPolicy creates a new retry policy

func (*RetryPolicy) CalculateDelay

func (rp *RetryPolicy) CalculateDelay(attempt int) time.Duration

CalculateDelay calculates the delay for a given attempt with jitter

func (*RetryPolicy) Execute

func (rp *RetryPolicy) Execute(ctx context.Context, operation func() error) error

Execute runs a function with retry logic (exponential backoff with jitter)

func (*RetryPolicy) GetMetrics

func (rp *RetryPolicy) GetMetrics() RetryMetrics

GetMetrics returns retry metrics

func (*RetryPolicy) IsRetryable

func (rp *RetryPolicy) IsRetryable(err error) bool

IsRetryable determines if an error should be retried

type Semaphore

type Semaphore struct {
	// contains filtered or unexported fields
}

Semaphore implements a counting semaphore

func NewSemaphore

func NewSemaphore(max int) *Semaphore

NewSemaphore creates a new semaphore

func (*Semaphore) Acquire

func (s *Semaphore) Acquire(timeout time.Duration) bool

Acquire tries to acquire a slot, blocking until available or timeout

func (*Semaphore) GetStats

func (s *Semaphore) GetStats() (current, max int, acquired, released, rejected, timeouts int64, avgHoldTime float64)

GetStats returns current semaphore stats

func (*Semaphore) Release

func (s *Semaphore) Release(holdTime time.Duration)

Release releases a slot

func (*Semaphore) SetMax

func (s *Semaphore) SetMax(max int)

SetMax updates the maximum concurrent slots

type StreamChunkPayload

type StreamChunkPayload struct {
	RequestID string          `json:"request_id"`
	Chunk     json.RawMessage `json:"chunk"` // OpenAI-compatible SSE chunk data
	Done      bool            `json:"done"`  // True when stream is complete
}

StreamChunkPayload represents a streaming chunk sent to Swan Inference

type StreamEndPayload

type StreamEndPayload struct {
	RequestID    string `json:"request_id"`
	Latency      int64  `json:"latency_ms"`
	TokensInput  int64  `json:"tokens_input,omitempty"`
	TokensOutput int64  `json:"tokens_output,omitempty"`
	StatusCode   int    `json:"status_code,omitempty"` // HTTP status code for error responses
	Error        string `json:"error,omitempty"`
}

StreamEndPayload signals end of stream with usage stats

type StreamResult

type StreamResult struct {
	TokensInput  int64
	TokensOutput int64
	Error        error
}

StreamResult contains the final result of a streaming inference including token usage

type StreamingInferenceHandler

type StreamingInferenceHandler func(requestID string, payload InferencePayload, sendChunk func(chunk []byte, done bool) error) *StreamResult

StreamingInferenceHandler handles streaming inference requests It receives a callback to send chunks back to Swan Inference and returns token usage

type TokenBucket

type TokenBucket struct {
	// contains filtered or unexported fields
}

TokenBucket implements a token bucket rate limiter

func NewTokenBucket

func NewTokenBucket(tokensPerSecond float64, burstSize int) *TokenBucket

NewTokenBucket creates a new token bucket

func (*TokenBucket) Allow

func (tb *TokenBucket) Allow() bool

Allow checks if a request is allowed and consumes a token

func (*TokenBucket) GetStats

func (tb *TokenBucket) GetStats() (tokens float64, rate float64, allowed, throttled int64)

GetStats returns current bucket stats

func (*TokenBucket) SetRate

func (tb *TokenBucket) SetRate(tokensPerSecond float64)

SetRate updates the token refill rate

type VerifyPayload

type VerifyPayload struct {
	ChallengeID   string          `json:"challenge_id"`
	ChallengeType string          `json:"challenge_type"`
	ModelID       string          `json:"model_id"`
	Challenge     json.RawMessage `json:"challenge"`
}

VerifyPayload is sent to provider for model verification

type VerifyResponsePayload

type VerifyResponsePayload struct {
	ChallengeID string          `json:"challenge_id"`
	Success     bool            `json:"success"`
	Response    json.RawMessage `json:"response"`
	Error       string          `json:"error,omitempty"`
}

VerifyResponsePayload is returned after processing a verification challenge

type WarmupHandler

type WarmupHandler func(payload WarmupPayload) (*WarmupResponse, error)

WarmupHandler handles model warmup requests

type WarmupPayload

type WarmupPayload struct {
	ModelID    string `json:"model_id"`
	WarmupType string `json:"warmup_type"` // "load" or "inference"
}

WarmupPayload is sent from Swan Inference to pre-load a model

type WarmupResponse

type WarmupResponse struct {
	RequestID  string `json:"request_id"`
	ModelID    string `json:"model_id"`
	Success    bool   `json:"success"`
	LoadTimeMs int64  `json:"load_time_ms,omitempty"`
	MemoryMB   int64  `json:"memory_mb,omitempty"`
	Error      string `json:"error,omitempty"`
}

WarmupResponse is returned by provider after warmup

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL