Files
LocalAI/core/config/distributed_config.go
T
mudler-agentandEttore Di Giacinto c7ce0c602c fix: stop listing the models directory on request paths and cache GGUF header reads (#12683)
* fix(http): stop listing the models directory on embeddings and bearer requests

Every embeddings request read the whole models directory before the body
was parsed, even when the body named its model. On a network-backed
models directory this cost seconds per request.

Embeddings now binds the body first and looks up a default model only
when the body has none, like chat. The pre-bind default lookup and the
gpt-4o constant default are gone from that route.

The first-available default lookup skips the directory when the filter
cannot accept an unconfigured file. Every usecase filter rejects files
without a config, so the scan never added results for them. Match-all,
nil and name filters keep the scan.

The bearer token check is now an exact test: a configured name, or one
stat of that name in the models directory. The value is a literal, not a
regular expression, and a name that is not a local relative path never
matches a file.

Assisted-by: Claude Code:claude-haiku-5-5

* perf(config): cache GGUF header reads for model defaults

Model defaults re-parsed the GGUF header on every SetDefaults call, which
runs per request and at startup reload. Put a bounded LRU in front of the
parser, keyed by absolute path and validated by size and modification
time. The cached value is a snapshot of the fields the defaults use, not
the parser output.

A changed size or mtime is a miss. Missing files and parse errors are not
cached, so the existing fallback is unchanged. LOCALAI_DISABLE_GUESSING,
the empty-path guard, the alias early return, the panic recovery and the
context-size default are all kept in front of the cache.

Assisted-by: Claude:claude-haiku-5-5

* test(http): cover listing-free default lookup on every usecase route

Assisted-by: Claude Code:claude-haiku-5-5

* feat(config): persist the GGUF header cache per instance

The header cache kept parsed GGUF results only in memory, so every restart
parsed each model file again before its defaults were known. Write the cache
to a per-instance file in the data path and load it at startup. A restarted
frontend with a fixed instance ID starts warm.

The file is versioned and records the estimator inputs (gguf-parser version,
candidate contexts, auto context size, offload count). A mismatch on any of
them makes the file ignored. Each entry keeps path, size and mtime, so a
changed model file is a miss on load. Writes run on one background saver,
debounced, through a temp file and rename. The entry bound applies to the
file as well. Without a data path the cache stays in memory only.

Documented as a per-instance exception in the model configuration docs.

Assisted-by: Claude:claude-haiku-5-5

* docs(embeddings): document model selection and the no-model error

Assisted-by: Claude Code:claude-haiku-5-5

* refactor(config): split the header cache into a generic cache

The GGUF header cache mixed two things: the mechanics (bounded LRU keyed
by path and identity, per-instance file, debounced atomic save, tolerant
load) and the GGUF fields. Move the mechanics into headerCache[S], driven
by a headerCacheSpec[S] that names the file prefix, schema, fingerprint,
parser and the snapshot encode/decode pair. The GGUF cache becomes one
instance of it.

File names, the identity check, LRU size, debounce, atomic write and the
corruption handling are unchanged. The on-disk JSON keeps the "estimator"
key so existing files still load.

Assisted-by: Claude Code:claude-haiku-5-5

* feat(config): share one GGUF header cache file between frontends

Every frontend used to write its own header cache file, named after its
instance ID, so a header read by one frontend was never reused by another.
The cache now uses one file, header-cache.json, in the data path, for all
frontends and for standalone mode. The instance ID no longer changes the
file name.

Frontends share the file in three ways. A save first merges the file's
entries in, so one frontend's write does not drop another's entries. A
miss reloads the file when its modification time has changed, and a hit
never reads it. Writes go to a temp file that is renamed into place, so a
reader never sees a partial file.

Two frontends saving at the same moment can still drop each other's last
entries. That costs one re-parse, never a wrong value, because every
entry is checked against its model file before use.

The shared file is valid only when every frontend mounts the same data
volume. The distributed-mode and model-configuration docs state this mode
and its volume limit.

Assisted-by: Claude:claude-haiku-5-5

* feat(config): give an instance ID its own header cache file

Assisted-by: Claude Code:claude-haiku-5-5

* style(config): gofmt the distributed config struct

Assisted-by: Claude Code:claude-haiku-5-5

---------

Co-authored-by: Ettore Di Giacinto <mudler@localai.io>
2026-10-14 01:06:14 +02:00

600 lines
25 KiB
Go

package config
import (
"cmp"
"fmt"
"time"
"github.com/mudler/LocalAI/core/services/messaging"
"github.com/mudler/LocalAI/pkg/natsauth"
"github.com/mudler/xlog"
)
// DistributedConfig holds configuration for horizontal scaling mode.
// When Enabled is true, PostgreSQL and NATS are required.
type DistributedConfig struct {
Enabled bool // --distributed / LOCALAI_DISTRIBUTED
InstanceID string // --instance-id / LOCALAI_INSTANCE_ID (auto-generated UUID if empty)
// InstanceIDGenerated is set when the ID above was generated at startup. A generated
// ID changes on every start, so on-disk state must not be keyed by it.
InstanceIDGenerated bool
NatsURL string // --nats-url / LOCALAI_NATS_URL
StorageURL string // --storage-url / LOCALAI_STORAGE_URL (S3 endpoint)
RegistrationToken string // --registration-token / LOCALAI_REGISTRATION_TOKEN (required token for node registration)
// RegistrationRequireAuth fails startup when distributed mode is enabled but
// RegistrationToken is empty. The default (false) keeps the historical
// fail-open behavior with a loud warning; production should set it so the
// node-register endpoints and the worker file-transfer server cannot run
// unauthenticated. Mirrors NatsRequireAuth for the NATS bus.
RegistrationRequireAuth bool // LOCALAI_REGISTRATION_REQUIRE_AUTH
// RequireAuth is the umbrella switch (LOCALAI_DISTRIBUTED_REQUIRE_AUTH) for
// distributed-mode auth: when true it implies BOTH NatsRequireAuth and
// RegistrationRequireAuth, so a single knob locks down the bus and the
// registration/file-transfer layer together. The granular flags remain
// available to enforce just one layer.
RequireAuth bool // LOCALAI_DISTRIBUTED_REQUIRE_AUTH
AutoApproveNodes bool // --auto-approve-nodes / LOCALAI_AUTO_APPROVE_NODES (skip admin approval for new workers)
// SharedModels asserts that every node (frontend and workers) mounts the
// SAME models directory at the SAME path (e.g. a shared volume, as in
// docker-compose.distributed.yaml). When true, the router skips staging
// model files to workers entirely: the frontend's absolute model paths are
// already valid on the worker, so re-uploading them into a per-model
// subdirectory only re-downloads what is already present (#10556). Default
// false preserves the historical per-node staging behavior.
SharedModels bool // --distributed-shared-models / LOCALAI_DISTRIBUTED_SHARED_MODELS
// NATS JWT auth (optional; see pkg/natsauth and docs/features/distributed-mode.md)
NatsAccountSeed string // LOCALAI_NATS_ACCOUNT_SEED — account signing seed to mint per-node worker JWTs
NatsServiceJWT string // LOCALAI_NATS_SERVICE_JWT — user JWT for frontends / agent workers
NatsServiceSeed string // LOCALAI_NATS_SERVICE_SEED — signing seed paired with service JWT
NatsWorkerJWTTTL time.Duration // LOCALAI_NATS_WORKER_JWT_TTL — minted worker JWT lifetime (default 24h)
NatsRequireAuth bool // LOCALAI_NATS_REQUIRE_AUTH — fail startup if NATS credentials are missing
NatsTLSCA string // LOCALAI_NATS_TLS_CA — PEM file for private CA (server verify)
NatsTLSCert string // LOCALAI_NATS_TLS_CERT — client cert for NATS mTLS
NatsTLSKey string // LOCALAI_NATS_TLS_KEY — client key paired with NatsTLSCert
// S3 configuration (used when StorageURL is set)
StorageBucket string // --storage-bucket / LOCALAI_STORAGE_BUCKET
StorageRegion string // --storage-region / LOCALAI_STORAGE_REGION
StorageAccessKey string // --storage-access-key / LOCALAI_STORAGE_ACCESS_KEY
StorageSecretKey string // --storage-secret-key / LOCALAI_STORAGE_SECRET_KEY
// Timeout configuration (all have sensible defaults — zero means use default)
MCPToolTimeout time.Duration // MCP tool execution timeout (default 360s)
MCPDiscoveryTimeout time.Duration // MCP discovery timeout (default 60s)
WorkerWaitTimeout time.Duration // Max wait for healthy worker at startup (default 5m)
DrainTimeout time.Duration // Time to wait for in-flight requests during drain (default 30s)
HealthCheckInterval time.Duration // Health monitor check interval (default 15s)
StaleNodeThreshold time.Duration // Time before a node is considered stale (default 5m)
NodeHeartbeatCheckpoint time.Duration // Minimum gap between durable heartbeat writes (default 60s, 0 = every beat)
// ModelConfigResyncInterval is how often a frontend compares its model
// configs with the shared models directory, to catch a change whose
// invalidation event it missed (default 30s).
ModelConfigResyncInterval time.Duration
// DisablePerModelHealthCheck turns off the health monitor's per-model
// gRPC probe. When enabled (the default), the monitor pings each model's
// gRPC address and removes stale node_models rows whose backend has
// crashed even though the worker's node-level heartbeat is still arriving.
// Without per-model probing, /embeddings and /completions can be dispatched
// to a backend that silently returns garbage (see also the cascading
// model-row cleanup on MarkUnhealthy / MarkDraining).
DisablePerModelHealthCheck bool
MCPCIJobTimeout time.Duration // MCP CI job execution timeout (default 10m)
BackendInstallTimeout time.Duration // NATS round-trip timeout for backend.install (default 15m)
BackendUpgradeTimeout time.Duration // NATS round-trip timeout for backend.upgrade (default 15m)
// ModelLoadTimeout is the gRPC deadline for the remote LoadModel call the
// router issues once a worker has the backend installed and the model files
// staged. It therefore covers only the backend's own checkpoint load and
// pipeline init, which for a multi-tens-of-GB diffusion/video checkpoint on
// unified memory can far exceed the 5m default.
ModelLoadTimeout time.Duration // gRPC deadline for remote LoadModel (default 5m)
// ModelLoadWait bounds how long an inference request waits for a model that
// is being cold-loaded before it is answered with 503 plus live progress. A
// held HTTP request cannot survive real infrastructure — an ingress or LB
// idle timeout kills a 20-minute request regardless of what LocalAI does —
// so the wait is bounded by default.
//
// Zero means unset (DefaultModelLoadWait applies); ModelLoadWaitUnbounded
// records the operator asking for unbounded waiting with
// LOCALAI_MODEL_LOAD_WAIT=0.
ModelLoadWait time.Duration
MaxUploadSize int64 // Maximum upload body size in bytes (default 50 GB)
AgentWorkerConcurrency int `yaml:"agent_worker_concurrency" json:"agent_worker_concurrency" env:"LOCALAI_AGENT_WORKER_CONCURRENCY"`
// DiskHeadroomDisabled turns off the scheduler's free-disk admission check,
// restoring the pre-#11054 behaviour where node selection ignores whether a
// node can actually store the model. The check is ON by default because it
// prevents a measured failure (a node with 0 bytes free accepted a 70GB
// model and failed 16 minutes into staging); this is the escape hatch for
// setups where our size estimate is wrong (deduplicating filesystems, a
// worker that fetches its own weights), not the norm.
//
// Disabling does NOT silence the check: it still runs and warns when it
// would have rejected every node, so the operator keeps the diagnosis
// without being blocked. See SmartRouter.scheduleNewModel.
//
// Stored as the negation of the CLI/runtime flag so the zero value is
// "enabled" (mirrors PrefixCacheDisabled).
DiskHeadroomDisabled bool
// PrefixCacheDisabled turns off prefix-cache-aware routing, falling back to
// round-robin (the floor). Prefix-cache routing is ON by default in
// distributed mode; this flag exists so operators can opt out. The CLI
// surfaces a default-true --distributed-prefix-cache enable flag and sets
// this when the operator passes --distributed-prefix-cache=false.
PrefixCacheDisabled bool
// PrefixCacheTTL is the idle-timeout for prefix-cache index entries and
// drives the background eviction cadence (eviction runs every TTL/2). Zero
// means use the prefixcache package default (5m).
PrefixCacheTTL time.Duration
// ModelSchedulingJSON is an inline JSON list of per-model scheduling configs
// applied authoritatively at startup (LOCALAI_MODEL_SCHEDULING).
ModelSchedulingJSON string
// ModelSchedulingConfigPath is a path to a YAML file with the same list
// (LOCALAI_MODEL_SCHEDULING_CONFIG).
ModelSchedulingConfigPath string
}
// Validate checks that the distributed configuration is internally consistent.
// It returns nil if distributed mode is disabled.
func (c DistributedConfig) Validate() error {
if !c.Enabled {
return nil
}
if c.NatsURL == "" {
return fmt.Errorf("distributed mode requires --nats-url / LOCALAI_NATS_URL")
}
// S3 credentials must be paired
if (c.StorageAccessKey != "" && c.StorageSecretKey == "") ||
(c.StorageAccessKey == "" && c.StorageSecretKey != "") {
return fmt.Errorf("storage-access-key and storage-secret-key must both be set or both empty")
}
// The registration token guards both the node HTTP register/heartbeat
// endpoints and the worker file-transfer server (which fails open on an
// empty token). Enforce it when registration auth is required (the granular
// flag or the umbrella); otherwise warn.
if c.RegistrationToken == "" {
if c.RegistrationAuthRequired() {
return fmt.Errorf("registration auth is required (LOCALAI_REGISTRATION_REQUIRE_AUTH or LOCALAI_DISTRIBUTED_REQUIRE_AUTH) but LOCALAI_REGISTRATION_TOKEN is empty")
}
xlog.Warn("distributed mode running without registration token — node endpoints and the worker file-transfer server are unprotected; set LOCALAI_REGISTRATION_TOKEN, or LOCALAI_DISTRIBUTED_REQUIRE_AUTH=true to fail closed")
}
if err := c.NatsAuthConfig().Validate(); err != nil {
return err
}
if err := c.NatsTLSFiles().Validate(); err != nil {
return err
}
c.NatsAuthConfig().WarnIfInsecure(true)
// Check for negative durations
for name, d := range map[string]time.Duration{
FlagMCPToolTimeout: c.MCPToolTimeout,
FlagMCPDiscoveryTimeout: c.MCPDiscoveryTimeout,
FlagWorkerWaitTimeout: c.WorkerWaitTimeout,
FlagDrainTimeout: c.DrainTimeout,
FlagHealthCheckInterval: c.HealthCheckInterval,
FlagStaleNodeThreshold: c.StaleNodeThreshold,
FlagNodeHeartbeatCheckpoint: c.NodeHeartbeatCheckpoint,
FlagMCPCIJobTimeout: c.MCPCIJobTimeout,
FlagBackendInstallTimeout: c.BackendInstallTimeout,
FlagBackendUpgradeTimeout: c.BackendUpgradeTimeout,
FlagModelLoadTimeout: c.ModelLoadTimeout,
} {
if d < 0 {
return fmt.Errorf("%s must not be negative", name)
}
}
if c.ModelConfigResyncInterval < 0 {
return fmt.Errorf("%s must not be negative", FlagModelConfigResyncInterval)
}
return nil
}
// Distributed config options
var EnableDistributed = func(o *ApplicationConfig) {
o.Distributed.Enabled = true
}
func WithDistributedInstanceID(id string) AppOption {
return func(o *ApplicationConfig) {
o.Distributed.InstanceID = id
}
}
func WithNatsURL(url string) AppOption {
return func(o *ApplicationConfig) {
o.Distributed.NatsURL = url
}
}
func WithRegistrationToken(token string) AppOption {
return func(o *ApplicationConfig) {
o.Distributed.RegistrationToken = token
}
}
func WithNatsAccountSeed(seed string) AppOption {
return func(o *ApplicationConfig) {
o.Distributed.NatsAccountSeed = seed
}
}
func WithNatsServiceJWT(jwt string) AppOption {
return func(o *ApplicationConfig) {
o.Distributed.NatsServiceJWT = jwt
}
}
func WithNatsServiceSeed(seed string) AppOption {
return func(o *ApplicationConfig) {
o.Distributed.NatsServiceSeed = seed
}
}
func WithNatsWorkerJWTTTL(d time.Duration) AppOption {
return func(o *ApplicationConfig) {
o.Distributed.NatsWorkerJWTTTL = d
}
}
var EnableNatsRequireAuth = func(o *ApplicationConfig) {
o.Distributed.NatsRequireAuth = true
}
// EnableRegistrationRequireAuth makes an empty registration token a hard error
// in distributed mode (see DistributedConfig.RegistrationRequireAuth).
var EnableRegistrationRequireAuth = func(o *ApplicationConfig) {
o.Distributed.RegistrationRequireAuth = true
}
// EnableDistributedRequireAuth is the umbrella switch implying both
// NatsRequireAuth and RegistrationRequireAuth (see DistributedConfig.RequireAuth).
var EnableDistributedRequireAuth = func(o *ApplicationConfig) {
o.Distributed.RequireAuth = true
}
// RegistrationAuthRequired reports whether an empty registration token must be
// treated as a fatal misconfiguration — the granular flag or the umbrella.
func (c DistributedConfig) RegistrationAuthRequired() bool {
return c.RegistrationRequireAuth || c.RequireAuth
}
// NatsAuthRequired reports whether NATS JWT credentials must be present — the
// granular flag or the umbrella.
func (c DistributedConfig) NatsAuthRequired() bool {
return c.NatsRequireAuth || c.RequireAuth
}
func WithNatsTLSCA(path string) AppOption {
return func(o *ApplicationConfig) {
o.Distributed.NatsTLSCA = path
}
}
func WithNatsTLSCert(path string) AppOption {
return func(o *ApplicationConfig) {
o.Distributed.NatsTLSCert = path
}
}
func WithNatsTLSKey(path string) AppOption {
return func(o *ApplicationConfig) {
o.Distributed.NatsTLSKey = path
}
}
func WithStorageURL(url string) AppOption {
return func(o *ApplicationConfig) {
o.Distributed.StorageURL = url
}
}
func WithStorageBucket(bucket string) AppOption {
return func(o *ApplicationConfig) {
o.Distributed.StorageBucket = bucket
}
}
func WithStorageRegion(region string) AppOption {
return func(o *ApplicationConfig) {
o.Distributed.StorageRegion = region
}
}
func WithStorageAccessKey(key string) AppOption {
return func(o *ApplicationConfig) {
o.Distributed.StorageAccessKey = key
}
}
func WithStorageSecretKey(key string) AppOption {
return func(o *ApplicationConfig) {
o.Distributed.StorageSecretKey = key
}
}
func WithBackendInstallTimeout(d time.Duration) AppOption {
return func(o *ApplicationConfig) {
o.Distributed.BackendInstallTimeout = d
}
}
func WithBackendUpgradeTimeout(d time.Duration) AppOption {
return func(o *ApplicationConfig) {
o.Distributed.BackendUpgradeTimeout = d
}
}
func WithModelLoadTimeout(d time.Duration) AppOption {
return func(o *ApplicationConfig) {
o.Distributed.ModelLoadTimeout = d
}
}
// WithModelLoadWait sets how long a request waits for a cold-loading model. A
// zero d records the operator asking for unbounded waiting: "set the knob to
// zero" cannot sensibly mean "use the default".
func WithModelLoadWait(d time.Duration) AppOption {
return func(o *ApplicationConfig) {
if d == 0 {
d = ModelLoadWaitUnbounded
}
o.Distributed.ModelLoadWait = d
}
}
// WithModelConfigResyncInterval sets how often a frontend resyncs its model
// configs from the shared models directory. Zero means the default.
func WithModelConfigResyncInterval(d time.Duration) AppOption {
return func(o *ApplicationConfig) {
o.Distributed.ModelConfigResyncInterval = d
}
}
// WithStaleNodeThreshold sets how long a node may go without a durable
// heartbeat before the health monitor marks it offline. It has to be raised
// alongside WithNodeHeartbeatCheckpoint: a checkpoint interval wider than this
// threshold makes every healthy node look dead the moment its beats start
// being suppressed.
func WithStaleNodeThreshold(d time.Duration) AppOption {
return func(o *ApplicationConfig) {
o.Distributed.StaleNodeThreshold = d
}
}
// WithNodeHeartbeatCheckpoint bounds durable heartbeat writes. A zero d is
// deliberately not special-cased into "unbounded": NodeHeartbeatCheckpointOrDefault
// reads zero as unset, and an operator who wants a write per beat sets a value
// below the worker's heartbeat interval instead.
func WithNodeHeartbeatCheckpoint(d time.Duration) AppOption {
return func(o *ApplicationConfig) {
o.Distributed.NodeHeartbeatCheckpoint = d
}
}
var EnableAutoApproveNodes = func(o *ApplicationConfig) {
o.Distributed.AutoApproveNodes = true
}
// EnableDistributedSharedModels marks the cluster as sharing one models
// directory across all nodes, so the router skips staging model files to
// workers (see DistributedConfig.SharedModels).
var EnableDistributedSharedModels = func(o *ApplicationConfig) {
o.Distributed.SharedModels = true
}
// DisableDiskHeadroomCheck turns off the scheduler's free-disk admission
// check (see DistributedConfig.DiskHeadroomDisabled). The check is enabled by
// default in distributed mode.
var DisableDiskHeadroomCheck = func(o *ApplicationConfig) {
o.Distributed.DiskHeadroomDisabled = true
}
// DisablePrefixCache turns off prefix-cache-aware routing (falls back to
// round-robin). Prefix-cache routing is enabled by default in distributed mode.
var DisablePrefixCache = func(o *ApplicationConfig) {
o.Distributed.PrefixCacheDisabled = true
}
// WithPrefixCacheTTL sets the prefix-cache index idle-timeout (and the
// background eviction cadence, which runs every TTL/2).
func WithPrefixCacheTTL(d time.Duration) AppOption {
return func(o *ApplicationConfig) {
o.Distributed.PrefixCacheTTL = d
}
}
// WithModelSchedulingJSON sets the inline-JSON declarative scheduling config.
func WithModelSchedulingJSON(s string) AppOption {
return func(o *ApplicationConfig) {
o.Distributed.ModelSchedulingJSON = s
}
}
// WithModelSchedulingConfigPath sets the path to a YAML declarative scheduling
// config file.
func WithModelSchedulingConfigPath(path string) AppOption {
return func(o *ApplicationConfig) {
o.Distributed.ModelSchedulingConfigPath = path
}
}
// Flag names for distributed timeout / interval configuration. These are
// the kebab-case identifiers kong derives from the matching RunCMD struct
// fields; they appear in Validate error messages and any other operator-
// facing surface that needs to reference a specific knob by name. Keeping
// them as constants prevents the string from drifting from the actual
// flag a future rename would produce.
const (
FlagMCPToolTimeout = "mcp-tool-timeout"
FlagMCPDiscoveryTimeout = "mcp-discovery-timeout"
FlagWorkerWaitTimeout = "worker-wait-timeout"
FlagDrainTimeout = "drain-timeout"
FlagHealthCheckInterval = "health-check-interval"
FlagStaleNodeThreshold = "stale-node-threshold"
FlagNodeHeartbeatCheckpoint = "node-heartbeat-checkpoint"
FlagMCPCIJobTimeout = "mcp-ci-job-timeout"
FlagBackendInstallTimeout = "backend-install-timeout"
FlagBackendUpgradeTimeout = "backend-upgrade-timeout"
FlagModelLoadTimeout = "model-load-timeout"
FlagModelLoadWait = "model-load-wait"
// FlagDiskHeadroomCheck names the disk-headroom toggle. It is quoted in
// the warning the check emits while disabled, so the operator reading a
// log line knows exactly which knob produced it.
FlagDiskHeadroomCheck = "distributed-disk-headroom-check"
)
// FlagModelConfigResyncInterval names the model config resync interval.
const FlagModelConfigResyncInterval = "model-config-resync-interval"
// DefaultModelConfigResyncInterval bounds how long a frontend that missed a
// models invalidation serves an old model config. It matches the tick of the
// replica reconciler, which acts on those configs.
const DefaultModelConfigResyncInterval = 30 * time.Second
// Defaults for distributed timeouts.
const (
DefaultMCPToolTimeout = 360 * time.Second
DefaultMCPDiscoveryTimeout = 60 * time.Second
DefaultWorkerWaitTimeout = 5 * time.Minute
DefaultDrainTimeout = 30 * time.Second
DefaultHealthCheckInterval = 15 * time.Second
// A beat that only refreshes the timestamp is now dropped until the
// checkpoint interval elapses, so the persisted column is up to one
// interval stale by design. The threshold covers that plus jitter.
// A genuinely dead node is still caught sooner by the per-model gRPC
// health check and by request-time failure, neither of which reads this.
DefaultStaleNodeThreshold = 5 * time.Minute
DefaultNodeHeartbeatCheckpoint = 60 * time.Second
DefaultMCPCIJobTimeout = 10 * time.Minute
DefaultBackendInstallTimeout = 15 * time.Minute
DefaultBackendUpgradeTimeout = 15 * time.Minute
DefaultModelLoadTimeout = 5 * time.Minute
// DefaultModelLoadWait is how long a request waits for a cold-loading model
// before it is answered with 503 and live progress. Chosen to sit under the
// idle timeout of typical ingress/LB defaults, so the answer comes from
// LocalAI (with progress the client can act on) rather than from a proxy
// dropping the connection.
DefaultModelLoadWait = 60 * time.Second
)
// ModelLoadWaitUnbounded records LOCALAI_MODEL_LOAD_WAIT=0 — "wait as long as
// it takes" — which a plain zero cannot express, since zero also means "unset,
// use the default". Only deployments with no proxy in front should use it.
const ModelLoadWaitUnbounded = -1 * time.Second
// DefaultMaxUploadSize is the default maximum upload body size (50 GB).
const DefaultMaxUploadSize int64 = 50 << 30
// NatsTLSFiles returns NATS TLS/mTLS PEM paths for the messaging client.
func (c DistributedConfig) NatsTLSFiles() messaging.TLSFiles {
return messaging.TLSFiles{
CA: c.NatsTLSCA,
Cert: c.NatsTLSCert,
Key: c.NatsTLSKey,
}
}
// NatsMessagingOptions builds messaging client options (JWT + TLS) for distributed components.
// Pass explicit userJWT/userSeed when set (e.g. worker overrides); empty uses service JWT from config.
func (c DistributedConfig) NatsMessagingOptions(userJWT, userSeed string) []messaging.Option {
var opts []messaging.Option
jwt, seed := userJWT, userSeed
if jwt == "" && seed == "" {
auth := c.NatsAuthConfig()
jwt, seed = auth.ServiceUserJWT, auth.ServiceUserSeed
}
if jwt != "" && seed != "" {
opts = append(opts, messaging.WithUserJWT(jwt, seed))
}
if tls := c.NatsTLSFiles(); tls.Enabled() {
opts = append(opts, messaging.WithTLS(tls))
}
return opts
}
// NatsAuthConfig builds pkg/natsauth settings from distributed configuration.
func (c DistributedConfig) NatsAuthConfig() natsauth.Config {
return natsauth.Config{
AccountSeed: c.NatsAccountSeed,
ServiceUserJWT: c.NatsServiceJWT,
ServiceUserSeed: c.NatsServiceSeed,
WorkerJWTTTL: c.NatsWorkerJWTTTL,
RequireAuth: c.NatsAuthRequired(),
}
}
// BackendInstallTimeoutOrDefault returns the configured timeout or the default.
func (c DistributedConfig) BackendInstallTimeoutOrDefault() time.Duration {
return cmp.Or(c.BackendInstallTimeout, DefaultBackendInstallTimeout)
}
// BackendUpgradeTimeoutOrDefault returns the configured timeout or the default.
func (c DistributedConfig) BackendUpgradeTimeoutOrDefault() time.Duration {
return cmp.Or(c.BackendUpgradeTimeout, DefaultBackendUpgradeTimeout)
}
// ModelLoadTimeoutOrDefault returns the configured timeout or the default.
func (c DistributedConfig) ModelLoadTimeoutOrDefault() time.Duration {
return cmp.Or(c.ModelLoadTimeout, DefaultModelLoadTimeout)
}
// MCPToolTimeoutOrDefault returns the configured timeout or the default.
func (c DistributedConfig) MCPToolTimeoutOrDefault() time.Duration {
return cmp.Or(c.MCPToolTimeout, DefaultMCPToolTimeout)
}
// MCPDiscoveryTimeoutOrDefault returns the configured timeout or the default.
func (c DistributedConfig) MCPDiscoveryTimeoutOrDefault() time.Duration {
return cmp.Or(c.MCPDiscoveryTimeout, DefaultMCPDiscoveryTimeout)
}
// WorkerWaitTimeoutOrDefault returns the configured timeout or the default.
func (c DistributedConfig) WorkerWaitTimeoutOrDefault() time.Duration {
return cmp.Or(c.WorkerWaitTimeout, DefaultWorkerWaitTimeout)
}
// DrainTimeoutOrDefault returns the configured timeout or the default.
func (c DistributedConfig) DrainTimeoutOrDefault() time.Duration {
return cmp.Or(c.DrainTimeout, DefaultDrainTimeout)
}
// HealthCheckIntervalOrDefault returns the configured interval or the default.
func (c DistributedConfig) HealthCheckIntervalOrDefault() time.Duration {
return cmp.Or(c.HealthCheckInterval, DefaultHealthCheckInterval)
}
// StaleNodeThresholdOrDefault returns the configured threshold or the default.
// ModelConfigResyncIntervalOrDefault returns the configured interval or the
// default.
func (c DistributedConfig) ModelConfigResyncIntervalOrDefault() time.Duration {
return cmp.Or(c.ModelConfigResyncInterval, DefaultModelConfigResyncInterval)
}
func (c DistributedConfig) StaleNodeThresholdOrDefault() time.Duration {
return cmp.Or(c.StaleNodeThreshold, DefaultStaleNodeThreshold)
}
// NodeHeartbeatCheckpointOrDefault returns the configured interval or the
// default. A configured zero is indistinguishable from unset here, which is
// intentional: cmp.Or falls back to the default, and an operator who wants a
// write per beat sets a value below the heartbeat interval instead.
func (c DistributedConfig) NodeHeartbeatCheckpointOrDefault() time.Duration {
return cmp.Or(c.NodeHeartbeatCheckpoint, DefaultNodeHeartbeatCheckpoint)
}
// MCPCIJobTimeoutOrDefault returns the configured MCP CI job timeout or the default.
func (c DistributedConfig) MCPCIJobTimeoutOrDefault() time.Duration {
return cmp.Or(c.MCPCIJobTimeout, DefaultMCPCIJobTimeout)
}
// MaxUploadSizeOrDefault returns the configured max upload size or the default.
func (c DistributedConfig) MaxUploadSizeOrDefault() int64 {
return cmp.Or(c.MaxUploadSize, DefaultMaxUploadSize)
}