cluster

package
v1.0.2 Latest Latest
Warning

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

Go to latest
Published: May 11, 2026 License: MIT Imports: 32 Imported by: 0

Documentation

Index

Constants

View Source
const DefaultConfigPath = "cluster.yaml"

Variables

This section is empty.

Functions

func APIKeyHash

func APIKeyHash(apiKey string) string

func ApplyOriginalAuthFileName

func ApplyOriginalAuthFileName(auths []*coreauth.Auth, filename string)

ApplyOriginalAuthFileName stores the source OAuth file name for management display.

func AutoMigrate

func AutoMigrate(db *gorm.DB) error

func Bootstrap

func Bootstrap(ctx context.Context, opts BootstrapOptions) error

func ClientAddr

func ClientAddr(ctx context.Context, db *gorm.DB) (string, error)

func ConfigRootFromSnapshot

func ConfigRootFromSnapshot(snapshot map[string]json.RawMessage) (map[string]any, error)

func DeterministicAPIKeyUUID

func DeterministicAPIKeyUUID(provider, baseURL, apiKeyHash, compatName, providerKey string) string

func DeterministicVirtualUUID

func DeterministicVirtualUUID(parentUUID, projectID string) string

func EnsureOAuthPayloadUUID

func EnsureOAuthPayloadUUID(raw []byte) ([]byte, string, bool, error)

func ForwardRefreshToMaster

func ForwardRefreshToMaster(ctx context.Context, master *ClusterNodeRecord, authUUID string, secret string) ([]byte, error)

func NormalizeClusterAuth

func NormalizeClusterAuth(auth *coreauth.Auth, uuidValue string)

func Open

func Open(ctx context.Context, cfg PGSQLConfig) (*gorm.DB, error)

func RecordToAuth

func RecordToAuth(record *AuthRecord) (*coreauth.Auth, error)

func RuntimeConfigFromRoot

func RuntimeConfigFromRoot(root map[string]any) (*appconfig.Config, []byte, error)

Types

type AuthIndex

type AuthIndex struct {
	UUID        string
	ID          string
	Index       string
	Provider    string
	Label       string
	Prefix      string
	Status      coreauth.Status
	Disabled    bool
	Unavailable bool
	BaseURL     string
	ModelsHash  string
	Attributes  map[string]string
}

type AuthRecord

type AuthRecord struct {
	UUID             string          `gorm:"column:uuid;primaryKey;type:uuid"`
	AuthJSON         JSONB           `gorm:"column:auth_json;type:jsonb;not null"`
	Version          int64           `gorm:"column:version;not null;default:1"`
	ID               string          `gorm:"column:id;index:idx_auth_active_order,priority:2"`
	Index            string          `gorm:"column:index"`
	Provider         string          `gorm:"column:provider"`
	Label            string          `gorm:"column:label"`
	Prefix           string          `gorm:"column:prefix"`
	Status           coreauth.Status `gorm:"column:status"`
	Disabled         bool            `gorm:"column:disabled"`
	Unavailable      bool            `gorm:"column:unavailable"`
	BaseURL          string          `gorm:"column:base_url"`
	APIKeyHash       string          `gorm:"column:api_key_hash"`
	CompatName       string          `gorm:"column:compat_name"`
	ProviderKey      string          `gorm:"column:provider_key"`
	ModelsHash       string          `gorm:"column:models_hash"`
	CreatedAt        time.Time       `gorm:"column:created_at"`
	UpdatedAt        time.Time       `gorm:"column:updated_at"`
	LastRefreshedAt  *time.Time      `gorm:"column:last_refreshed_at"`
	NextRefreshAfter *time.Time      `gorm:"column:next_refresh_after"`
	DeletedAt        gorm.DeletedAt  `gorm:"column:deleted_at;index;index:idx_auth_active_order,priority:1"`
}

func AuthToRecord

func AuthToRecord(auth *coreauth.Auth) (*AuthRecord, error)

func (AuthRecord) TableName

func (AuthRecord) TableName() string

type BootstrapOptions

type BootstrapOptions struct {
	ConfigPath string
	AuthDir    string
	Config     *appconfig.Config
	Repository *Repository
	Now        time.Time
}

type ClusterEventRecord

type ClusterEventRecord struct {
	ID         uint      `gorm:"column:id;primaryKey;autoIncrement"`
	Scope      string    `gorm:"column:scope"`
	Op         string    `gorm:"column:op"`
	EntityUUID string    `gorm:"column:entity_uuid"`
	Version    int64     `gorm:"column:version"`
	CreatedAt  time.Time `gorm:"column:created_at"`
}

func (ClusterEventRecord) TableName

func (ClusterEventRecord) TableName() string

type ClusterNodeRecord

type ClusterNodeRecord struct {
	IP         string    `` /* 150-byte string literal not displayed */
	Port       int       `` /* 152-byte string literal not displayed */
	SecretHash string    `gorm:"column:secret_hash;index:idx_cluster_auth_lookup,priority:2"`
	IsMaster   bool      `gorm:"column:is_master;index:idx_cluster_master_nodes,priority:1"`
	StartedAt  time.Time `` /* 147-byte string literal not displayed */
	LastSeenAt time.Time `` /* 149-byte string literal not displayed */
}

func (ClusterNodeRecord) TableName

func (ClusterNodeRecord) TableName() string

type Config

type Config struct {
	PGSQL PGSQLConfig `yaml:"pgsql"`
	Node  NodeConfig  `yaml:"node"`
}

func LoadConfigOptional

func LoadConfigOptional(path string) (*Config, bool, error)

func (*Config) Validate

func (c *Config) Validate() error

type ConfigRecord

type ConfigRecord struct {
	Key       string    `gorm:"column:key;primaryKey"`
	Value     JSONB     `gorm:"column:value;type:jsonb;not null"`
	Version   int64     `gorm:"column:version;not null;default:1"`
	CreatedAt time.Time `gorm:"column:created_at"`
	UpdatedAt time.Time `gorm:"column:updated_at"`
}

func (ConfigRecord) TableName

func (ConfigRecord) TableName() string

type Coordinator

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

func NewCoordinator

func NewCoordinator(repo *Repository, node NodeIdentity, opts CoordinatorOptions) *Coordinator

func (*Coordinator) CurrentMaster

func (c *Coordinator) CurrentMaster(ctx context.Context) (*ClusterNodeRecord, error)

func (*Coordinator) IsMaster

func (c *Coordinator) IsMaster() bool

func (*Coordinator) NodeSecret

func (c *Coordinator) NodeSecret() string

func (*Coordinator) SetOnMasterChanged

func (c *Coordinator) SetOnMasterChanged(callback func(bool))

func (*Coordinator) Start

func (c *Coordinator) Start(ctx context.Context) error

type CoordinatorOptions

type CoordinatorOptions struct {
	HeartbeatInterval time.Duration
	HeartbeatTimeout  time.Duration
	OnMasterChanged   func(bool)
}

type DisbandOptions

type DisbandOptions struct {
	ConfigPath string
	AuthDir    string
	Repository *Repository
}

type DisbandResult

type DisbandResult struct {
	ConfigPath          string
	AuthDir             string
	ConfigBytes         int
	AuthFiles           int
	GeminiKeys          int
	VertexKeys          int
	CodexKeys           int
	ClaudeKeys          int
	OpenAICompatibility int
}

func Disband

func Disband(ctx context.Context, opts DisbandOptions) (*DisbandResult, error)

type EventWatcher

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

func NewEventWatcher

func NewEventWatcher(repo *Repository, pollInterval time.Duration, onEvent func(context.Context, ClusterEventRecord) error) *EventWatcher

func NewEventWatcherFrom

func NewEventWatcherFrom(repo *Repository, pollInterval time.Duration, lastSeenID int64, onEvent func(context.Context, ClusterEventRecord) error) *EventWatcher

func (*EventWatcher) Start

func (w *EventWatcher) Start(ctx context.Context) error

type JSONB

type JSONB []byte

func (JSONB) MarshalJSON

func (j JSONB) MarshalJSON() ([]byte, error)

func (*JSONB) Scan

func (j *JSONB) Scan(value any) error

func (*JSONB) UnmarshalJSON

func (j *JSONB) UnmarshalJSON(data []byte) error

func (JSONB) Value

func (j JSONB) Value() (driver.Value, error)

type NodeConfig

type NodeConfig struct {
	Port              int           `yaml:"port"`
	HeartbeatInterval time.Duration `yaml:"heartbeat-interval"`
	HeartbeatTimeout  time.Duration `yaml:"heartbeat-timeout"`
	EventPollInterval time.Duration `yaml:"event-poll-interval"`
}

type NodeIdentity

type NodeIdentity struct {
	IP        string
	Port      int
	Secret    string
	StartedAt time.Time
}

type PGSQLConfig

type PGSQLConfig struct {
	Host     string `yaml:"host"`
	Port     int    `yaml:"port"`
	User     string `yaml:"user"`
	Password string `yaml:"password"`
	Passowrd string `yaml:"passowrd"`
	Database string `yaml:"database"`
	SSLMode  string `yaml:"sslmode"`
}

func (PGSQLConfig) DSN

func (c PGSQLConfig) DSN() (string, error)

func (PGSQLConfig) Validate

func (c PGSQLConfig) Validate() error

type RESPHandler

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

func NewRESPHandler

func NewRESPHandler(coordinator *Coordinator, refresh *RefreshController, repo *Repository) *RESPHandler

func (*RESPHandler) Handle

func (h *RESPHandler) Handle(ctx context.Context, args []string, remoteIP string) ([]byte, error)

type RefreshController

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

func NewRefreshController

func NewRefreshController(coordinator *Coordinator, runtime *home.Runtime, repo *Repository) *RefreshController

func (*RefreshController) CanAutoRefresh

func (c *RefreshController) CanAutoRefresh() bool

func (*RefreshController) OnMasterChanged

func (c *RefreshController) OnMasterChanged(isMaster bool)

func (*RefreshController) RefreshNow

func (c *RefreshController) RefreshNow(ctx context.Context, authIndex string) ([]byte, error)

type Repository

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

func NewRepository

func NewRepository(db *gorm.DB) *Repository

func (*Repository) AppendEvent

func (r *Repository) AppendEvent(ctx context.Context, scope, op, entity string, version int64) error

func (*Repository) AppendUsage

func (r *Repository) AppendUsage(ctx context.Context, payload string) (*UsageRecord, error)

func (*Repository) GetAuth

func (r *Repository) GetAuth(ctx context.Context, uuid string) (*coreauth.Auth, *AuthRecord, error)

func (*Repository) ListAuthIndex

func (r *Repository) ListAuthIndex(ctx context.Context) ([]AuthIndex, error)

func (*Repository) ListAuths

func (r *Repository) ListAuths(ctx context.Context) ([]*coreauth.Auth, error)

func (*Repository) LiveClusterNodeByIPAndSecret

func (r *Repository) LiveClusterNodeByIPAndSecret(ctx context.Context, ip string, secret string, cutoff time.Time) (*ClusterNodeRecord, error)

func (*Repository) LoadConfigAsRuntimeConfig

func (r *Repository) LoadConfigAsRuntimeConfig(ctx context.Context) (*appconfig.Config, []byte, error)

func (*Repository) LoadConfigSnapshot

func (r *Repository) LoadConfigSnapshot(ctx context.Context) (map[string]json.RawMessage, error)

func (*Repository) MaxEventID

func (r *Repository) MaxEventID(ctx context.Context) (int64, error)

func (*Repository) ReplaceConfigSnapshot

func (r *Repository) ReplaceConfigSnapshot(ctx context.Context, values map[string]any) error

func (*Repository) SoftDeleteAuth

func (r *Repository) SoftDeleteAuth(ctx context.Context, uuid string) error

func (*Repository) UpsertAuth

func (r *Repository) UpsertAuth(ctx context.Context, auth *coreauth.Auth, op string) (*AuthRecord, error)

func (*Repository) UpsertConfigValue

func (r *Repository) UpsertConfigValue(ctx context.Context, key string, value any) error

func (*Repository) WithAuthRefreshLock

func (r *Repository) WithAuthRefreshLock(ctx context.Context, uuid string, fn func(tx *Repository, auth *coreauth.Auth) (*coreauth.Auth, error)) (*coreauth.Auth, error)

type RuntimeAdapter

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

func NewRuntimeAdapter

func NewRuntimeAdapter(repo *Repository) *RuntimeAdapter

func (*RuntimeAdapter) ApplyEvent

func (a *RuntimeAdapter) ApplyEvent(ctx context.Context, event ClusterEventRecord) error

func (*RuntimeAdapter) Delete

func (a *RuntimeAdapter) Delete(ctx context.Context, id string) error

func (*RuntimeAdapter) Enabled

func (a *RuntimeAdapter) Enabled() bool

func (*RuntimeAdapter) GetFullAuth

func (a *RuntimeAdapter) GetFullAuth(ctx context.Context, uuid string) (*coreauth.Auth, error)

func (*RuntimeAdapter) InvalidateFullAuth

func (a *RuntimeAdapter) InvalidateFullAuth(uuid string)

func (*RuntimeAdapter) List

func (a *RuntimeAdapter) List(ctx context.Context) ([]*coreauth.Auth, error)

func (*RuntimeAdapter) ListMinimalAuths

func (a *RuntimeAdapter) ListMinimalAuths() []*coreauth.Auth

func (*RuntimeAdapter) LoadAuthIndex

func (a *RuntimeAdapter) LoadAuthIndex(ctx context.Context) error

func (*RuntimeAdapter) LoadConfigYAML

func (a *RuntimeAdapter) LoadConfigYAML(ctx context.Context) ([]byte, error)

func (*RuntimeAdapter) LoadIndex

func (a *RuntimeAdapter) LoadIndex(ctx context.Context) error

func (*RuntimeAdapter) RefreshAuthIndex

func (a *RuntimeAdapter) RefreshAuthIndex(ctx context.Context, uuid string) error

func (*RuntimeAdapter) RemoveAuthIndex

func (a *RuntimeAdapter) RemoveAuthIndex(uuid string)

func (*RuntimeAdapter) Save

func (a *RuntimeAdapter) Save(ctx context.Context, auth *coreauth.Auth) (string, error)

func (*RuntimeAdapter) StoreUsagePayload

func (a *RuntimeAdapter) StoreUsagePayload(ctx context.Context, payload string) error

type UsageRecord

type UsageRecord struct {
	ID              uint      `gorm:"column:id;primaryKey;autoIncrement"`
	Timestamp       time.Time `gorm:"column:timestamp;type:timestamptz;not null;index:idx_usage_timestamp"`
	LatencyMS       int64     `gorm:"column:latency_ms;not null;default:0"`
	Source          string    `gorm:"column:source;index:idx_usage_source"`
	AuthIndex       string    `gorm:"column:auth_index;index:idx_usage_auth_index"`
	InputTokens     int64     `gorm:"column:input_tokens;not null;default:0"`
	OutputTokens    int64     `gorm:"column:output_tokens;not null;default:0"`
	ReasoningTokens int64     `gorm:"column:reasoning_tokens;not null;default:0"`
	CachedTokens    int64     `gorm:"column:cached_tokens;not null;default:0"`
	TotalTokens     int64     `gorm:"column:total_tokens;not null;default:0"`
	Failed          bool      `gorm:"column:failed;not null;default:false;index:idx_usage_failed"`
	FailStatusCode  int       `gorm:"column:fail_status_code;not null;default:0"`
	FailBody        string    `gorm:"column:fail_body;type:text"`
	Provider        string    `gorm:"column:provider;index:idx_usage_provider_model,priority:1"`
	Model           string    `gorm:"column:model;index:idx_usage_provider_model,priority:2"`
	Alias           string    `gorm:"column:alias"`
	Endpoint        string    `gorm:"column:endpoint;index:idx_usage_endpoint"`
	AuthType        string    `gorm:"column:auth_type"`
	APIKey          string    `gorm:"column:api_key"`
	RequestID       string    `gorm:"column:request_id;index:idx_usage_request_id"`
	TokensJSON      JSONB     `gorm:"column:tokens;type:jsonb"`
	FailJSON        JSONB     `gorm:"column:fail;type:jsonb"`
	PayloadJSON     JSONB     `gorm:"column:payload;type:jsonb;not null"`
	CreatedAt       time.Time `gorm:"column:created_at;type:timestamptz;not null"`
}

func UsageRecordFromPayload

func UsageRecordFromPayload(payload string) (*UsageRecord, error)

func (UsageRecord) TableName

func (UsageRecord) TableName() string

Directories

Path Synopsis

Jump to

Keyboard shortcuts

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