From 6584db992f406bfb0b29019b792be03087faeac8 Mon Sep 17 00:00:00 2001 From: "mudler's LocalAI [bot]" <139863280+localai-bot@users.noreply.github.com> Date: Thu, 23 Jul 2026 00:03:21 +0200 Subject: [PATCH] fix(nodes): never schedule a model onto a node that cannot store it (#11054) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(nodes): never schedule a model onto a node that cannot store it A worker whose models filesystem was 100% full kept advertising `status: healthy`, stayed a scheduling candidate, was picked to host a 70 GB video model, accepted the staging request, transferred ~17 GB and only then failed: staging .../whisper-large-v3/model.fp32-00001-of-00002.safetensors: upload to node b7bacbf4-... failed with status 500: writing file: /models/longcat-video-avatar-1.5/...: no space left on device The node was at 937G/937G/0-avail. Total elapsed before the truth surfaced: 16 minutes, for a decision that could never have succeeded. The worker health signal only ever proved liveness. `/readyz` (WorkerReadiness/NATSReadiness) checks the NATS link; `status: healthy` in the registry is driven by heartbeat recency. Node capacity carried VRAM and RAM but no disk figure at all, and the router compared model size against VRAM only — nothing anywhere looked at free space on the filesystem that staging actually writes to. Report it, then use it: - Workers now measure the filesystem backing their MODELS directory (not `/` -- staged weights land in the models path, and that mount is very often separate) and report `total_disk`/`available_disk` on registration and on every heartbeat. Free disk moves faster than VRAM under staging traffic, so the per-heartbeat refresh matters. - The SmartRouter drops nodes that cannot store the model before it picks one. The requirement comes from `modelPayloadBytes` -- the same local paths `stageModelFiles` uploads, already computed for the size-derived load budget -- plus a 5% / 1 GiB margin, rather than a fixed percentage of the node's disk. A percentage threshold would take a small-but-usable node out of rotation for models it could hold, and on a homogeneous cluster would strand every node at once. - When no node fits, scheduling fails immediately with an error naming the requirement and each node's free space, instead of picking one and discovering it mid-transfer. Two deliberate non-changes. Low disk does not mark a node `unhealthy`: the check is per model, so a node too small for one model stays a valid target for smaller ones. And `total_disk == 0` means "does not report disk" (pre-upgrade worker, or a failed stat), not "full" -- such nodes pass through untouched so a rolling upgrade never empties the candidate pool. A genuinely full node is distinguishable: non-zero total, zero available. Registry read failures are logged and scheduling continues unfiltered; a database hiccup must not wedge a cluster. Free space is surfaced on the node detail page next to VRAM, since the incident's signature was a node that looked entirely healthy. Signed-off-by: Ettore Di Giacinto Assisted-by: Claude Code:claude-opus-4-8[1m] [Read] [Edit] [Bash] * feat(nodes): make the disk-headroom check operator-controllable The admission check added in the previous commit had no off switch. A scheduler-side veto with no escape hatch is a liability: our size estimate can be wrong (deduplicating or compressing filesystems, a backend that fetches its own weights rather than loading the staged copy), and an operator who hits that has no way out but a downgrade. Add one knob with two surfaces that share a single source of truth: - `--distributed-disk-headroom-check` / `LOCALAI_DISTRIBUTED_DISK_HEADROOM_CHECK` (default true), following the `--distributed-prefix-cache` pattern for a default-on distributed feature. - `distributed_disk_headroom_check` in the runtime-settings registry, so it can be flipped without a restart from `POST /api/settings` and from Settings -> Distributed in the WebUI. Both write `DistributedConfig.DiskHeadroomDisabled`, and the SmartRouter reads that member LIVE on every scheduling decision through a closure over the application config rather than a value snapshotted at construction. Env/CLI sets the boot value, the runtime setting overrides it live, last write wins, and there is exactly one member to read. Snapshotting would have made the runtime toggle a no-op until restart. Disabled means WARN, not SKIP. Selection goes back to ignoring free disk -- byte for byte the pre-check behaviour -- but the check still runs, and when it would have rejected every node it says so, naming the knob that suppressed it. Going quiet when switched off would reproduce the exact condition that made the original incident expensive: a cluster doing something that could not work and saying nothing. Disabling is also logged once at startup. Warning only on the total-rejection case keeps it actionable rather than chatty on a heterogeneous cluster. Also fixes a false positive in the check itself: shared-models mode (LOCALAI_DISTRIBUTED_SHARED_MODELS) stages nothing at all -- every node already mounts this models directory at this path -- so demanding the full checkpoint size of free space per node would have rejected a cluster that needs no new bytes. The check is skipped there entirely. Signed-off-by: Ettore Di Giacinto Assisted-by: Claude Code:claude-opus-4-8[1m] [Read] [Edit] [Bash] --------- Signed-off-by: Ettore Di Giacinto Co-authored-by: Ettore Di Giacinto --- core/application/distributed.go | 13 + core/cli/run.go | 62 +++-- core/config/disk_headroom_setting_test.go | 51 ++++ core/config/distributed_config.go | 27 ++ core/config/runtime_settings.go | 6 + core/config/runtime_settings_registry.go | 10 + ...runtime_settings_registry_internal_test.go | 1 + core/http/endpoints/localai/nodes.go | 11 +- .../react-ui/public/locales/de/admin.json | 1 + .../react-ui/public/locales/en/admin.json | 1 + .../react-ui/public/locales/es/admin.json | 1 + .../react-ui/public/locales/id/admin.json | 1 + .../react-ui/public/locales/it/admin.json | 1 + .../react-ui/public/locales/ko/admin.json | 1 + .../react-ui/public/locales/zh-CN/admin.json | 1 + core/http/react-ui/src/pages/NodeDetail.jsx | 10 + core/http/react-ui/src/pages/Settings.jsx | 13 + core/services/nodes/disk_headroom.go | 140 ++++++++++ core/services/nodes/disk_headroom_test.go | 255 ++++++++++++++++++ core/services/nodes/interfaces.go | 1 + core/services/nodes/model_router_test.go | 3 + core/services/nodes/registry.go | 45 +++- core/services/nodes/router.go | 105 ++++++-- core/services/nodes/router_test.go | 18 ++ core/services/worker/registration.go | 26 ++ docs/content/features/distributed-mode.md | 47 +++- pkg/xsysinfo/disk.go | 68 +++++ pkg/xsysinfo/disk_test.go | 36 +++ 28 files changed, 901 insertions(+), 54 deletions(-) create mode 100644 core/config/disk_headroom_setting_test.go create mode 100644 core/services/nodes/disk_headroom.go create mode 100644 core/services/nodes/disk_headroom_test.go create mode 100644 pkg/xsysinfo/disk.go create mode 100644 pkg/xsysinfo/disk_test.go diff --git a/core/application/distributed.go b/core/application/distributed.go index 3cef85bc9..750b73b02 100644 --- a/core/application/distributed.go +++ b/core/application/distributed.go @@ -356,6 +356,12 @@ func initDistributed(cfg *config.ApplicationConfig, authDB *gorm.DB, configLoade PrefixConfig: prefixCfg, Pressure: pressure, SharedModels: cfg.Distributed.SharedModels, + // A closure over the live ApplicationConfig, NOT a snapshot: the + // runtime setting (distributed_disk_headroom_check) mutates this exact + // member, so a snapshot here would make the toggle a no-op until + // restart. env/CLI sets the boot value, POST /api/settings overrides it + // live, and this is the single member both write. + DiskHeadroomEnabled: func() bool { return !cfg.Distributed.DiskHeadroomDisabled }, // RAW, not OrDefault: zero means "derive the budget per model from the // checkpoint size" (config.ModelLoadTimeoutForSize), which is what makes // a 70 GB video checkpoint work without the operator first hitting a @@ -377,6 +383,13 @@ func initDistributed(cfg *config.ApplicationConfig, authDB *gorm.DB, configLoade // replica, not just the one performing the transfer. Without this, a // /api/operations poll that round-robins onto a peer sees no staging row and // the progress flickers. The origin publishes; peers mirror via the wildcard. + // A silently disabled safety check is how the original incident stayed + // invisible for sixteen minutes. Say so once, loudly, at startup. + if cfg.Distributed.DiskHeadroomDisabled { + xlog.Info("Disk-headroom admission check is DISABLED: node selection will ignore whether a worker can store the model, and staging may fail with ENOSPC partway through a transfer", + "knob", config.FlagDiskHeadroomCheck, "env", "LOCALAI_DISTRIBUTED_DISK_HEADROOM_CHECK") + } + router.StagingTracker().SetPublisher(natsClient) if _, err := router.StagingTracker().SubscribeBroadcasts(natsClient); err != nil { xlog.Warn("Failed to subscribe to staging progress broadcasts", "error", err) diff --git a/core/cli/run.go b/core/cli/run.go index 77a9f23e6..600b2a997 100644 --- a/core/cli/run.go +++ b/core/cli/run.go @@ -158,35 +158,36 @@ type RunCMD struct { DefaultAPIKeyExpiry string `env:"LOCALAI_DEFAULT_API_KEY_EXPIRY" help:"Default expiry for API keys (e.g. 90d, 1y; empty = no expiry)" group:"auth"` // Distributed / Horizontal Scaling - Distributed bool `env:"LOCALAI_DISTRIBUTED" default:"false" help:"Enable distributed mode (requires PostgreSQL + NATS)" group:"distributed"` - InstanceID string `env:"LOCALAI_INSTANCE_ID" help:"Unique instance ID for distributed mode (auto-generated UUID if empty)" group:"distributed"` - NatsURL string `env:"LOCALAI_NATS_URL" help:"NATS server URL (e.g., nats://localhost:4222)" group:"distributed"` - StorageURL string `env:"LOCALAI_STORAGE_URL" help:"S3-compatible storage endpoint URL (e.g., http://minio:9000)" group:"distributed"` - StorageBucket string `env:"LOCALAI_STORAGE_BUCKET" default:"localai" help:"S3 bucket name for object storage" group:"distributed"` - StorageRegion string `env:"LOCALAI_STORAGE_REGION" default:"us-east-1" help:"S3 region" group:"distributed"` - StorageAccessKey string `env:"LOCALAI_STORAGE_ACCESS_KEY" help:"S3 access key ID" group:"distributed"` - StorageSecretKey string `env:"LOCALAI_STORAGE_SECRET_KEY" help:"S3 secret access key" group:"distributed"` - RegistrationToken string `env:"LOCALAI_REGISTRATION_TOKEN" help:"Token that backend nodes must provide to register (empty = no auth required)" group:"distributed"` - RegistrationRequireAuth bool `env:"LOCALAI_REGISTRATION_REQUIRE_AUTH" default:"false" help:"Fail startup when distributed mode is enabled but LOCALAI_REGISTRATION_TOKEN is empty (node endpoints and worker file-transfer server would otherwise be unauthenticated)" group:"distributed"` - DistributedRequireAuth bool `env:"LOCALAI_DISTRIBUTED_REQUIRE_AUTH" default:"false" help:"Umbrella switch: require BOTH NATS JWT credentials and a registration token when distributed mode is enabled (implies --nats-require-auth and --registration-require-auth)" group:"distributed"` - AutoApproveNodes bool `env:"LOCALAI_AUTO_APPROVE_NODES" default:"false" help:"Auto-approve new worker nodes (skip admin approval)" group:"distributed"` - DistributedSharedModels bool `env:"LOCALAI_DISTRIBUTED_SHARED_MODELS" default:"false" help:"Assert that every node mounts the SAME models directory at the SAME path (shared volume). When true, the router skips staging model files to workers and loads them directly from the shared path, avoiding re-downloads." group:"distributed"` - DistributedPrefixCache bool `env:"LOCALAI_DISTRIBUTED_PREFIX_CACHE" default:"true" help:"Enable prefix-cache-aware routing in distributed mode (default true). When false, routing falls back to round-robin." group:"distributed"` - DistributedPrefixCacheTTL string `env:"LOCALAI_DISTRIBUTED_PREFIX_CACHE_TTL" help:"Idle-timeout for prefix-cache index entries; also drives the background eviction cadence (every TTL/2). Default 5m." group:"distributed"` - BackendInstallTimeout string `env:"LOCALAI_NATS_BACKEND_INSTALL_TIMEOUT" help:"NATS round-trip timeout for backend.install requests sent to worker nodes (default 15m). Increase for slow links pulling multi-GB images." group:"distributed"` - BackendUpgradeTimeout string `env:"LOCALAI_NATS_BACKEND_UPGRADE_TIMEOUT" help:"NATS round-trip timeout for backend.upgrade requests (default 15m)." group:"distributed"` - ModelLoadTimeout string `env:"LOCALAI_NATS_MODEL_LOAD_TIMEOUT" help:"Fixed gRPC deadline for the remote LoadModel call sent to a worker node once its backend is installed and model files are staged. Unset (the default), the deadline is derived from the checkpoint size instead: 5m plus 20s per GiB, capped at 6h, so multi-tens-of-GB diffusion/video checkpoints get the minutes they need without a fixed cliff. Set this only to pin a specific budget; the value is used verbatim, including when it is shorter than the derived one." group:"distributed"` - NatsAccountSeed string `env:"LOCALAI_NATS_ACCOUNT_SEED" help:"NATS account signing seed (SU...) used to mint per-node worker JWTs at registration" group:"distributed"` - NatsServiceJWT string `env:"LOCALAI_NATS_SERVICE_JWT" help:"NATS user JWT for the frontend (and agent workers) to publish control-plane messages" group:"distributed"` - NatsServiceSeed string `env:"LOCALAI_NATS_SERVICE_SEED" help:"NATS user signing seed (SU...) paired with LOCALAI_NATS_SERVICE_JWT" group:"distributed"` - NatsWorkerJWTTTL string `env:"LOCALAI_NATS_WORKER_JWT_TTL" help:"Lifetime of minted per-node NATS JWTs (e.g. 24h, default 24h)" group:"distributed"` - NatsRequireAuth bool `env:"LOCALAI_NATS_REQUIRE_AUTH" default:"false" help:"Require NATS JWT credentials (service JWT + account seed) when distributed mode is enabled" group:"distributed"` - NatsTLSCA string `env:"LOCALAI_NATS_TLS_CA" type:"existingfile" help:"PEM file for NATS server CA (private PKI); use with tls:// in --nats-url" group:"distributed"` - NatsTLSCert string `env:"LOCALAI_NATS_TLS_CERT" type:"existingfile" help:"Client certificate for NATS mTLS" group:"distributed"` - NatsTLSKey string `env:"LOCALAI_NATS_TLS_KEY" type:"existingfile" help:"Client private key for NATS mTLS" group:"distributed"` - ExposeNodeHeader bool `env:"LOCALAI_EXPOSE_NODE_HEADER" default:"false" help:"Set the X-LocalAI-Node response header on inference responses (OpenAI chat/completions/embeddings, Anthropic /v1/messages, Ollama /api/chat,/api/generate,/api/embed) with the ID of the worker that served the request. Disabled by default: the node ID reveals internal topology and should not be exposed on a public endpoint. Best-effort: under heavy concurrency the header may reflect a recent routing decision rather than this exact request's." group:"distributed"` - ModelScheduling string `env:"LOCALAI_MODEL_SCHEDULING" help:"Declarative per-model scheduling config applied at startup (inline JSON list of {model_name,node_selector,min_replicas,max_replicas,replicas:\"all\"}). Authoritative: overwrites matching models on every boot. Distributed mode only." group:"distributed"` - ModelSchedulingConfig string `env:"LOCALAI_MODEL_SCHEDULING_CONFIG" help:"Path to a YAML file with the same per-model scheduling list as LOCALAI_MODEL_SCHEDULING. Distributed mode only." group:"distributed"` + Distributed bool `env:"LOCALAI_DISTRIBUTED" default:"false" help:"Enable distributed mode (requires PostgreSQL + NATS)" group:"distributed"` + InstanceID string `env:"LOCALAI_INSTANCE_ID" help:"Unique instance ID for distributed mode (auto-generated UUID if empty)" group:"distributed"` + NatsURL string `env:"LOCALAI_NATS_URL" help:"NATS server URL (e.g., nats://localhost:4222)" group:"distributed"` + StorageURL string `env:"LOCALAI_STORAGE_URL" help:"S3-compatible storage endpoint URL (e.g., http://minio:9000)" group:"distributed"` + StorageBucket string `env:"LOCALAI_STORAGE_BUCKET" default:"localai" help:"S3 bucket name for object storage" group:"distributed"` + StorageRegion string `env:"LOCALAI_STORAGE_REGION" default:"us-east-1" help:"S3 region" group:"distributed"` + StorageAccessKey string `env:"LOCALAI_STORAGE_ACCESS_KEY" help:"S3 access key ID" group:"distributed"` + StorageSecretKey string `env:"LOCALAI_STORAGE_SECRET_KEY" help:"S3 secret access key" group:"distributed"` + RegistrationToken string `env:"LOCALAI_REGISTRATION_TOKEN" help:"Token that backend nodes must provide to register (empty = no auth required)" group:"distributed"` + RegistrationRequireAuth bool `env:"LOCALAI_REGISTRATION_REQUIRE_AUTH" default:"false" help:"Fail startup when distributed mode is enabled but LOCALAI_REGISTRATION_TOKEN is empty (node endpoints and worker file-transfer server would otherwise be unauthenticated)" group:"distributed"` + DistributedRequireAuth bool `env:"LOCALAI_DISTRIBUTED_REQUIRE_AUTH" default:"false" help:"Umbrella switch: require BOTH NATS JWT credentials and a registration token when distributed mode is enabled (implies --nats-require-auth and --registration-require-auth)" group:"distributed"` + AutoApproveNodes bool `env:"LOCALAI_AUTO_APPROVE_NODES" default:"false" help:"Auto-approve new worker nodes (skip admin approval)" group:"distributed"` + DistributedSharedModels bool `env:"LOCALAI_DISTRIBUTED_SHARED_MODELS" default:"false" help:"Assert that every node mounts the SAME models directory at the SAME path (shared volume). When true, the router skips staging model files to workers and loads them directly from the shared path, avoiding re-downloads." group:"distributed"` + DistributedPrefixCache bool `env:"LOCALAI_DISTRIBUTED_PREFIX_CACHE" default:"true" help:"Enable prefix-cache-aware routing in distributed mode (default true). When false, routing falls back to round-robin." group:"distributed"` + DistributedDiskHeadroomCheck bool `env:"LOCALAI_DISTRIBUTED_DISK_HEADROOM_CHECK" default:"true" help:"Reject worker nodes that lack free space to store the model, at scheduling time rather than partway through staging (default true). Free space is measured on the filesystem backing each worker's models directory, and compared against the model's own size plus a small margin. When false, node selection ignores free disk (pre-#11054 behaviour); the check still runs and warns when it would have rejected every node. Can also be toggled at runtime via the distributed_disk_headroom_check setting." group:"distributed"` + DistributedPrefixCacheTTL string `env:"LOCALAI_DISTRIBUTED_PREFIX_CACHE_TTL" help:"Idle-timeout for prefix-cache index entries; also drives the background eviction cadence (every TTL/2). Default 5m." group:"distributed"` + BackendInstallTimeout string `env:"LOCALAI_NATS_BACKEND_INSTALL_TIMEOUT" help:"NATS round-trip timeout for backend.install requests sent to worker nodes (default 15m). Increase for slow links pulling multi-GB images." group:"distributed"` + BackendUpgradeTimeout string `env:"LOCALAI_NATS_BACKEND_UPGRADE_TIMEOUT" help:"NATS round-trip timeout for backend.upgrade requests (default 15m)." group:"distributed"` + ModelLoadTimeout string `env:"LOCALAI_NATS_MODEL_LOAD_TIMEOUT" help:"Fixed gRPC deadline for the remote LoadModel call sent to a worker node once its backend is installed and model files are staged. Unset (the default), the deadline is derived from the checkpoint size instead: 5m plus 20s per GiB, capped at 6h, so multi-tens-of-GB diffusion/video checkpoints get the minutes they need without a fixed cliff. Set this only to pin a specific budget; the value is used verbatim, including when it is shorter than the derived one." group:"distributed"` + NatsAccountSeed string `env:"LOCALAI_NATS_ACCOUNT_SEED" help:"NATS account signing seed (SU...) used to mint per-node worker JWTs at registration" group:"distributed"` + NatsServiceJWT string `env:"LOCALAI_NATS_SERVICE_JWT" help:"NATS user JWT for the frontend (and agent workers) to publish control-plane messages" group:"distributed"` + NatsServiceSeed string `env:"LOCALAI_NATS_SERVICE_SEED" help:"NATS user signing seed (SU...) paired with LOCALAI_NATS_SERVICE_JWT" group:"distributed"` + NatsWorkerJWTTTL string `env:"LOCALAI_NATS_WORKER_JWT_TTL" help:"Lifetime of minted per-node NATS JWTs (e.g. 24h, default 24h)" group:"distributed"` + NatsRequireAuth bool `env:"LOCALAI_NATS_REQUIRE_AUTH" default:"false" help:"Require NATS JWT credentials (service JWT + account seed) when distributed mode is enabled" group:"distributed"` + NatsTLSCA string `env:"LOCALAI_NATS_TLS_CA" type:"existingfile" help:"PEM file for NATS server CA (private PKI); use with tls:// in --nats-url" group:"distributed"` + NatsTLSCert string `env:"LOCALAI_NATS_TLS_CERT" type:"existingfile" help:"Client certificate for NATS mTLS" group:"distributed"` + NatsTLSKey string `env:"LOCALAI_NATS_TLS_KEY" type:"existingfile" help:"Client private key for NATS mTLS" group:"distributed"` + ExposeNodeHeader bool `env:"LOCALAI_EXPOSE_NODE_HEADER" default:"false" help:"Set the X-LocalAI-Node response header on inference responses (OpenAI chat/completions/embeddings, Anthropic /v1/messages, Ollama /api/chat,/api/generate,/api/embed) with the ID of the worker that served the request. Disabled by default: the node ID reveals internal topology and should not be exposed on a public endpoint. Best-effort: under heavy concurrency the header may reflect a recent routing decision rather than this exact request's." group:"distributed"` + ModelScheduling string `env:"LOCALAI_MODEL_SCHEDULING" help:"Declarative per-model scheduling config applied at startup (inline JSON list of {model_name,node_selector,min_replicas,max_replicas,replicas:\"all\"}). Authoritative: overwrites matching models on every boot. Distributed mode only." group:"distributed"` + ModelSchedulingConfig string `env:"LOCALAI_MODEL_SCHEDULING_CONFIG" help:"Path to a YAML file with the same per-model scheduling list as LOCALAI_MODEL_SCHEDULING. Distributed mode only." group:"distributed"` Version bool @@ -409,6 +410,9 @@ func (r *RunCMD) Run(ctx *cliContext.Context) error { if !r.DistributedPrefixCache { opts = append(opts, config.DisablePrefixCache) } + if !r.DistributedDiskHeadroomCheck { + opts = append(opts, config.DisableDiskHeadroomCheck) + } if r.DistributedPrefixCacheTTL != "" { d, err := time.ParseDuration(r.DistributedPrefixCacheTTL) if err != nil { diff --git a/core/config/disk_headroom_setting_test.go b/core/config/disk_headroom_setting_test.go new file mode 100644 index 000000000..62f8de1c7 --- /dev/null +++ b/core/config/disk_headroom_setting_test.go @@ -0,0 +1,51 @@ +package config_test + +import ( + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + + "github.com/mudler/LocalAI/core/config" +) + +// The disk-headroom admission check has two operator surfaces that must not +// drift into two sources of truth: an env/CLI flag that sets the boot value, +// and a runtime setting that can override it live. Both write the SAME +// ApplicationConfig member, which the SmartRouter reads on every scheduling +// decision. +var _ = Describe("distributed disk-headroom check setting", func() { + It("is enabled by default", func() { + o := config.NewApplicationConfig() + + Expect(o.Distributed.DiskHeadroomDisabled).To(BeFalse(), + "the check must default ON — it prevents a measured production failure") + Expect(*o.ToRuntimeSettings().DistributedDiskHeadroomCheck).To(BeTrue()) + }) + + It("reports the env/CLI value through the runtime settings snapshot", func() { + o := config.NewApplicationConfig(config.DisableDiskHeadroomCheck) + + Expect(*o.ToRuntimeSettings().DistributedDiskHeadroomCheck).To(BeFalse()) + }) + + It("lets a runtime override beat the env value, in both directions", func() { + // Env said "off"; the operator turns it back on at runtime. + o := config.NewApplicationConfig(config.DisableDiskHeadroomCheck) + on := true + o.ApplyRuntimeSettings(&config.RuntimeSettings{DistributedDiskHeadroomCheck: &on}) + Expect(o.Distributed.DiskHeadroomDisabled).To(BeFalse()) + + // Env said "on" (the default); the operator turns it off at runtime. + o = config.NewApplicationConfig() + off := false + o.ApplyRuntimeSettings(&config.RuntimeSettings{DistributedDiskHeadroomCheck: &off}) + Expect(o.Distributed.DiskHeadroomDisabled).To(BeTrue()) + }) + + It("leaves the value alone when the runtime settings do not mention it", func() { + o := config.NewApplicationConfig(config.DisableDiskHeadroomCheck) + + o.ApplyRuntimeSettings(&config.RuntimeSettings{}) + + Expect(o.Distributed.DiskHeadroomDisabled).To(BeTrue()) + }) +}) diff --git a/core/config/distributed_config.go b/core/config/distributed_config.go index f486ba196..3cbbcf77b 100644 --- a/core/config/distributed_config.go +++ b/core/config/distributed_config.go @@ -88,6 +88,22 @@ type DistributedConfig struct { AgentWorkerConcurrency int `yaml:"agent_worker_concurrency" json:"agent_worker_concurrency" env:"LOCALAI_AGENT_WORKER_CONCURRENCY"` JobWorkerConcurrency int `yaml:"job_worker_concurrency" json:"job_worker_concurrency" env:"LOCALAI_JOB_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 @@ -310,6 +326,13 @@ 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) { @@ -356,6 +379,10 @@ const ( FlagBackendInstallTimeout = "backend-install-timeout" FlagBackendUpgradeTimeout = "backend-upgrade-timeout" FlagModelLoadTimeout = "model-load-timeout" + // 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" ) // Defaults for distributed timeouts. diff --git a/core/config/runtime_settings.go b/core/config/runtime_settings.go index 73c78ccda..7c29e6337 100644 --- a/core/config/runtime_settings.go +++ b/core/config/runtime_settings.go @@ -80,6 +80,12 @@ type RuntimeSettings struct { AgentPoolDatabaseURL *string `json:"agent_pool_database_url,omitempty"` // PostgreSQL DSN when vector engine is postgres AgentPoolAgentHubURL *string `json:"agent_pool_agent_hub_url,omitempty"` // override the agenthub.localai.io endpoint + // Distributed-mode settings. Read LIVE by the SmartRouter on each + // scheduling decision, so toggling takes effect on the next placement + // without a restart. Stored as the negation of + // DistributedConfig.DiskHeadroomDisabled for UI clarity. + DistributedDiskHeadroomCheck *bool `json:"distributed_disk_headroom_check,omitempty"` // Reject nodes without room to store the model (default: true) + // LocalAI Assistant settings — read live by the chat handler at request // entry, so flipping the toggle takes effect on the next request. LocalAIAssistantEnabled *bool `json:"localai_assistant_enabled,omitempty"` // negation of DisableLocalAIAssistant for UI clarity diff --git a/core/config/runtime_settings_registry.go b/core/config/runtime_settings_registry.go index fadb543a7..f06e14c91 100644 --- a/core/config/runtime_settings_registry.go +++ b/core/config/runtime_settings_registry.go @@ -418,6 +418,16 @@ var runtimeSettingsFields = []fieldSpec{ func(o *ApplicationConfig, v string) { o.AgentPool.AgentHubURL = v }, restartRequired()), + // Distributed disk-headroom admission check: stored as the negation for UI + // clarity (the config member is the "disabled" flag so its zero value is + // the safe, enabled default). No restartRequired(): the SmartRouter reads + // the ApplicationConfig member live on every scheduling decision, which is + // the whole point of exposing it here rather than as env/CLI only. + field("distributed_disk_headroom_check", + func(s *RuntimeSettings) **bool { return &s.DistributedDiskHeadroomCheck }, + func(o *ApplicationConfig) bool { return !o.Distributed.DiskHeadroomDisabled }, + func(o *ApplicationConfig, v bool) { o.Distributed.DiskHeadroomDisabled = !v }), + // LocalAI Assistant: stored as the negation for UI clarity. field("localai_assistant_enabled", func(s *RuntimeSettings) **bool { return &s.LocalAIAssistantEnabled }, diff --git a/core/config/runtime_settings_registry_internal_test.go b/core/config/runtime_settings_registry_internal_test.go index 14d93f126..7124eb85b 100644 --- a/core/config/runtime_settings_registry_internal_test.go +++ b/core/config/runtime_settings_registry_internal_test.go @@ -101,6 +101,7 @@ var _ = Describe("runtime settings registry", func() { src.AgentPool.DatabaseURL = "postgres://x" src.AgentPool.AgentHubURL = "https://hub" src.DisableLocalAIAssistant = true + src.Distributed.DiskHeadroomDisabled = true src.Branding = BrandingConfig{ InstanceName: "n", InstanceTagline: "t", LogoFile: "l.png", LogoHorizontalFile: "lh.png", FaviconFile: "f.ico", diff --git a/core/http/endpoints/localai/nodes.go b/core/http/endpoints/localai/nodes.go index bb9fc0225..bc26baf49 100644 --- a/core/http/endpoints/localai/nodes.go +++ b/core/http/endpoints/localai/nodes.go @@ -84,6 +84,12 @@ type RegisterNodeRequest struct { AvailableVRAM uint64 `json:"available_vram,omitempty"` TotalRAM uint64 `json:"total_ram,omitempty"` AvailableRAM uint64 `json:"available_ram,omitempty"` + // TotalDisk / AvailableDisk describe the filesystem backing the worker's + // MODELS directory (where staged weights land), not the root filesystem. + // Omitted by workers that predate the fields; the scheduler treats + // total_disk == 0 as "unknown" and leaves such a node in rotation. + TotalDisk uint64 `json:"total_disk,omitempty"` + AvailableDisk uint64 `json:"available_disk,omitempty"` GPUVendor string `json:"gpu_vendor,omitempty"` // GPUComputeCapability is the worker GPU's compute capability ("major.minor", // e.g. "12.1" for GB10). Used by the router for per-arch option tuning. @@ -176,6 +182,8 @@ func RegisterNodeEndpoint(registry *nodes.NodeRegistry, expectedToken string, au AvailableVRAM: req.AvailableVRAM, TotalRAM: req.TotalRAM, AvailableRAM: req.AvailableRAM, + TotalDisk: req.TotalDisk, + AvailableDisk: req.AvailableDisk, GPUVendor: req.GPUVendor, GPUComputeCapability: req.GPUComputeCapability, Capability: req.Capability, @@ -372,7 +380,8 @@ func HeartbeatEndpoint(registry *nodes.NodeRegistry) echo.HandlerFunc { _ = c.Bind(&update) // best-effort — empty body is fine var updatePtr *nodes.HeartbeatUpdate - if update.AvailableVRAM != nil || update.TotalVRAM != nil || update.AvailableRAM != nil || update.GPUVendor != "" { + if update.AvailableVRAM != nil || update.TotalVRAM != nil || update.AvailableRAM != nil || + update.AvailableDisk != nil || update.TotalDisk != nil || update.GPUVendor != "" { updatePtr = &update } diff --git a/core/http/react-ui/public/locales/de/admin.json b/core/http/react-ui/public/locales/de/admin.json index 3bf9daa68..a0d2f1226 100644 --- a/core/http/react-ui/public/locales/de/admin.json +++ b/core/http/react-ui/public/locales/de/admin.json @@ -23,6 +23,7 @@ "agents": "Agentenaufgaben", "agentpool": "Agenten-Pool", "assistant": "LocalAI Assistant", + "distributed": "Verteilt", "responses": "Antworten" } }, diff --git a/core/http/react-ui/public/locales/en/admin.json b/core/http/react-ui/public/locales/en/admin.json index 05155dd25..3dc860e1c 100644 --- a/core/http/react-ui/public/locales/en/admin.json +++ b/core/http/react-ui/public/locales/en/admin.json @@ -23,6 +23,7 @@ "agents": "Agent Jobs", "agentpool": "Agent Pool", "assistant": "LocalAI Assistant", + "distributed": "Distributed", "responses": "Responses" } }, diff --git a/core/http/react-ui/public/locales/es/admin.json b/core/http/react-ui/public/locales/es/admin.json index 1d4b61180..47177fd8b 100644 --- a/core/http/react-ui/public/locales/es/admin.json +++ b/core/http/react-ui/public/locales/es/admin.json @@ -23,6 +23,7 @@ "agents": "Trabajos de agentes", "agentpool": "Pool de agentes", "assistant": "LocalAI Assistant", + "distributed": "Distribuido", "responses": "Respuestas" } }, diff --git a/core/http/react-ui/public/locales/id/admin.json b/core/http/react-ui/public/locales/id/admin.json index 28fa5829c..7e9460f1d 100644 --- a/core/http/react-ui/public/locales/id/admin.json +++ b/core/http/react-ui/public/locales/id/admin.json @@ -23,6 +23,7 @@ "agents": "Agent Job", "agentpool": "Agent Pool", "assistant": "Asisten LocalAI", + "distributed": "Terdistribusi", "responses": "Respons" } }, diff --git a/core/http/react-ui/public/locales/it/admin.json b/core/http/react-ui/public/locales/it/admin.json index 323bae421..7d6a47523 100644 --- a/core/http/react-ui/public/locales/it/admin.json +++ b/core/http/react-ui/public/locales/it/admin.json @@ -23,6 +23,7 @@ "agents": "Job degli agenti", "agentpool": "Pool agenti", "assistant": "LocalAI Assistant", + "distributed": "Distribuito", "responses": "Risposte" } }, diff --git a/core/http/react-ui/public/locales/ko/admin.json b/core/http/react-ui/public/locales/ko/admin.json index 1b6676571..3a21b7c26 100644 --- a/core/http/react-ui/public/locales/ko/admin.json +++ b/core/http/react-ui/public/locales/ko/admin.json @@ -23,6 +23,7 @@ "agents": "에이전트 작업", "agentpool": "에이전트 풀", "assistant": "LocalAI 어시스턴트", + "distributed": "분산", "responses": "응답" } }, diff --git a/core/http/react-ui/public/locales/zh-CN/admin.json b/core/http/react-ui/public/locales/zh-CN/admin.json index c5d9db452..38a597ee0 100644 --- a/core/http/react-ui/public/locales/zh-CN/admin.json +++ b/core/http/react-ui/public/locales/zh-CN/admin.json @@ -23,6 +23,7 @@ "agents": "智能体任务", "agentpool": "智能体池", "assistant": "LocalAI Assistant", + "distributed": "分布式", "responses": "响应" } }, diff --git a/core/http/react-ui/src/pages/NodeDetail.jsx b/core/http/react-ui/src/pages/NodeDetail.jsx index 6414cb4c3..fb7ac1eec 100644 --- a/core/http/react-ui/src/pages/NodeDetail.jsx +++ b/core/http/react-ui/src/pages/NodeDetail.jsx @@ -96,6 +96,16 @@ export default function NodeDetail() { {formatVRAM(usedVRAM) || '0'} / {formatVRAM(node.total_vram)} )} + {node.total_disk > 0 && ( +
+ {/* Free space on the worker's MODELS filesystem. A node can look + perfectly healthy on VRAM while having nowhere to put the + weights, which is why this sits next to VRAM rather than + buried in a diagnostics panel. */} +
Models disk free
+ {formatVRAM(node.available_disk || 0) || '0'} / {formatVRAM(node.total_disk)} +
+ )}
In-flight
{node.in_flight_count || 0} diff --git a/core/http/react-ui/src/pages/Settings.jsx b/core/http/react-ui/src/pages/Settings.jsx index 2cf6abda2..5299454f6 100644 --- a/core/http/react-ui/src/pages/Settings.jsx +++ b/core/http/react-ui/src/pages/Settings.jsx @@ -26,6 +26,7 @@ const SECTIONS = [ { id: 'agents', icon: 'fa-tasks', color: 'var(--color-primary)' }, { id: 'agentpool', icon: 'fa-robot', color: 'var(--color-primary)' }, { id: 'assistant', icon: 'fa-user-shield', color: 'var(--color-accent)' }, + { id: 'distributed', icon: 'fa-server', color: 'var(--color-accent)' }, { id: 'responses', icon: 'fa-database', color: 'var(--color-accent)' }, ] @@ -640,6 +641,18 @@ export default function Settings() {
+ {/* Distributed mode */} +
sectionRefs.current.distributed = el} style={{ marginBottom: 'var(--spacing-xl)' }}> +

+ {t('settings.sections.distributed')} +

+
+ + update('distributed_disk_headroom_check', v)} /> + +
+
+ {/* Open Responses */}
sectionRefs.current.responses = el} style={{ marginBottom: 'var(--spacing-xl)' }}>

diff --git a/core/services/nodes/disk_headroom.go b/core/services/nodes/disk_headroom.go new file mode 100644 index 000000000..a93043725 --- /dev/null +++ b/core/services/nodes/disk_headroom.go @@ -0,0 +1,140 @@ +package nodes + +import ( + "context" + "errors" + "fmt" + "strings" + + "gorm.io/gorm" +) + +// ErrInsufficientDisk reports that no candidate node has enough free space on +// its models filesystem to store the model being scheduled. +// +// This is a scheduling-time verdict on purpose. Before it existed, a worker +// whose models filesystem was 100% full still advertised `status: healthy`, +// was picked to host a 70GB model, accepted the staging request, streamed +// ~17GB and only then failed with "no space left on device" — sixteen minutes +// after a decision that could never have succeeded. +var ErrInsufficientDisk = errors.New("no node has enough free disk for the model") + +const ( + // diskHeadroomMarginRatio is the fraction of the payload kept free on top + // of the payload itself. Staging is not the only writer on that + // filesystem (backend installs, logs, the backend's own scratch files), + // and the payload figure is a floor rather than an exact prediction. + diskHeadroomMarginRatio = 0.05 + + // diskHeadroomMinMarginBytes floors the proportional margin so small + // models still leave usable space behind them. + diskHeadroomMinMarginBytes = uint64(1) << 30 // 1 GiB + + // diskHeadroomUnknownSizeBytes is what we demand when the model's payload + // cannot be sized locally (a bare HuggingFace repo id the worker will + // fetch itself). Deliberately small: it is a "this filesystem is not + // wedged" floor, not a capacity estimate. Demanding more would strand + // small-but-healthy nodes on every model whose size we cannot see. + diskHeadroomUnknownSizeBytes = uint64(2) << 30 // 2 GiB +) + +// DiskRequirementFor returns the free bytes a node must have on its models +// filesystem before it may be handed a model whose staged payload is +// payloadBytes (as computed by modelPayloadBytes). +// +// The requirement is derived from the ACTUAL model size rather than from a +// fixed percentage of the node's disk. A percentage threshold would take a +// small node out of rotation for models it could comfortably hold, which on a +// homogeneous cluster strands every node at once. +func DiskRequirementFor(payloadBytes int64) uint64 { + if payloadBytes <= 0 { + return diskHeadroomUnknownSizeBytes + } + payload := uint64(payloadBytes) + margin := uint64(float64(payload) * diskHeadroomMarginRatio) + if margin < diskHeadroomMinMarginBytes { + margin = diskHeadroomMinMarginBytes + } + return payload + margin +} + +// reportsDisk says whether a node's disk figures are usable. +// +// TotalDisk is the "does this worker report disk at all" bit, NOT +// AvailableDisk: a 100%-full node legitimately reports available == 0, and +// treating that as unknown would reinstate exactly the bug this guards +// against. A worker predating the field (or one whose stat failed) reports +// total == 0 and is passed through untouched, so a rolling upgrade never +// empties the candidate pool. +func (n BackendNode) reportsDisk() bool { return n.TotalDisk > 0 } + +// nodesWithDiskHeadroom filters candidates down to those that can actually +// store a model needing `required` free bytes on their models filesystem. +func nodesWithDiskHeadroom(candidates []BackendNode, required uint64) []BackendNode { + fit := make([]BackendNode, 0, len(candidates)) + for _, n := range candidates { + if !n.reportsDisk() || n.AvailableDisk >= required { + fit = append(fit, n) + } + } + return fit +} + +// describeDiskShortfall renders the per-node free-space readings that produced +// a rejection, so the operator sees the numbers behind the decision instead of +// a bare "no nodes available". +func describeDiskShortfall(candidates []BackendNode) string { + parts := make([]string, 0, len(candidates)) + for _, n := range candidates { + parts = append(parts, fmt.Sprintf("%s has %s free of %s", + n.Name, humanFileSize(int64(n.AvailableDisk)), humanFileSize(int64(n.TotalDisk)))) + } + return strings.Join(parts, ", ") +} + +// NarrowByDiskHeadroom restricts candidateNodeIDs to nodes whose models +// filesystem can hold `required` more bytes. +// +// A nil candidateNodeIDs means "any healthy backend node" (the caller applied +// no selector); the returned slice is then the narrowed set, never nil, so the +// caller keeps the constraint. When nothing fits, the error wraps +// ErrInsufficientDisk and names every candidate's free space. +// +// Errors reading the registry are NOT fatal: the caller gets its original +// candidate set back alongside the error and can carry on. A database hiccup +// must not stop a cluster from scheduling. +func (r *NodeRegistry) NarrowByDiskHeadroom(ctx context.Context, candidateNodeIDs []string, required uint64) ([]string, error) { + candidates, err := r.healthyBackendNodes(ctx, candidateNodeIDs) + if err != nil { + return candidateNodeIDs, err + } + // No rows at all is not a disk verdict — the pool is empty for some other + // reason (nothing registered, everything unhealthy). Let the existing + // "no healthy nodes" paths report that; claiming a disk shortage here + // would be a misleading diagnosis. + if len(candidates) == 0 { + return candidateNodeIDs, nil + } + + fit := nodesWithDiskHeadroom(candidates, required) + if len(fit) == 0 { + return nil, fmt.Errorf("%w: need %s free on the models filesystem, but %s", + ErrInsufficientDisk, humanFileSize(int64(required)), describeDiskShortfall(candidates)) + } + return extractNodeIDs(fit), nil +} + +// healthyBackendNodes loads the healthy backend nodes, optionally restricted to +// an explicit candidate set. +func (r *NodeRegistry) healthyBackendNodes(ctx context.Context, candidateNodeIDs []string) ([]BackendNode, error) { + q := r.db.WithContext(ctx).Model(&BackendNode{}). + Where("status = ? AND node_type = ?", StatusHealthy, NodeTypeBackend) + if candidateNodeIDs != nil { + q = q.Where("id IN ?", candidateNodeIDs) + } + var out []BackendNode + if err := q.Find(&out).Error; err != nil && !errors.Is(err, gorm.ErrRecordNotFound) { + return nil, fmt.Errorf("listing healthy backend nodes: %w", err) + } + return out, nil +} diff --git a/core/services/nodes/disk_headroom_test.go b/core/services/nodes/disk_headroom_test.go new file mode 100644 index 000000000..217fb71fd --- /dev/null +++ b/core/services/nodes/disk_headroom_test.go @@ -0,0 +1,255 @@ +package nodes + +import ( + "context" + "errors" + "fmt" + "runtime" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + + "github.com/mudler/LocalAI/core/services/messaging" + "github.com/mudler/LocalAI/core/services/testutil" + pb "github.com/mudler/LocalAI/pkg/grpc/proto" + "gorm.io/gorm" +) + +// These specs encode the production incident behind this fix: a Jetson Thor +// worker whose models filesystem was 937G/937G/0-avail kept reporting +// `status: healthy`, was picked to host a 70GB model, accepted the staging +// request, transferred ~17GB and only then failed with +// "no space left on device" — 16 minutes after the decision that could never +// have worked. +var _ = Describe("Node disk headroom", func() { + const gb = uint64(1000 * 1000 * 1000) + + Describe("DiskRequirementFor", func() { + It("demands the model's payload plus a safety margin", func() { + // A 70GB checkpoint must require MORE than 70GB free: the staging + // write is not the only thing landing on that filesystem. + req := DiskRequirementFor(int64(70 * gb)) + Expect(req).To(BeNumerically(">", 70*gb)) + // ...but the margin must stay modest, or a tight-but-usable node + // gets stranded out of rotation. + Expect(req).To(BeNumerically("<", 80*gb)) + }) + + It("falls back to a small absolute floor when the model size is unknown", func() { + // Bare-HF-repo models have no local payload to stat. We must not + // demand 0 (which lets a completely full node back in) nor + // something large (which would strand a healthy small node). + req := DiskRequirementFor(0) + Expect(req).To(BeNumerically(">", 0)) + Expect(req).To(BeNumerically("<=", 4*gb)) + }) + }) + + Describe("nodesWithDiskHeadroom", func() { + It("drops a node whose models filesystem is full", func() { + full := BackendNode{ + Name: "nvidia-thor", TotalDisk: 937 * gb, AvailableDisk: 0, + } + roomy := BackendNode{ + Name: "worker-roomy", TotalDisk: 2000 * gb, AvailableDisk: 1200 * gb, + } + + fit := nodesWithDiskHeadroom([]BackendNode{full, roomy}, DiskRequirementFor(int64(70*gb))) + + Expect(nodeNames(fit)).To(ConsistOf("worker-roomy")) + }) + + It("drops a node that is merely too small for THIS model but keeps it for a smaller one", func() { + tight := BackendNode{Name: "worker-tight", TotalDisk: 100 * gb, AvailableDisk: 20 * gb} + + big := nodesWithDiskHeadroom([]BackendNode{tight}, DiskRequirementFor(int64(70*gb))) + Expect(big).To(BeEmpty()) + + small := nodesWithDiskHeadroom([]BackendNode{tight}, DiskRequirementFor(int64(2*gb))) + Expect(nodeNames(small)).To(ConsistOf("worker-tight")) + }) + + It("keeps nodes that do not report disk at all", func() { + // A pre-upgrade worker reports total_disk == 0. Excluding it would + // take the whole cluster out of rotation on a rolling upgrade. + legacy := BackendNode{Name: "worker-legacy", TotalDisk: 0, AvailableDisk: 0} + + fit := nodesWithDiskHeadroom([]BackendNode{legacy}, DiskRequirementFor(int64(70*gb))) + + Expect(nodeNames(fit)).To(ConsistOf("worker-legacy")) + }) + }) + + Describe("NarrowByDiskHeadroom", func() { + var ( + db *gorm.DB + registry *NodeRegistry + ) + + BeforeEach(func() { + if runtime.GOOS == "darwin" { + Skip("testcontainers requires Docker, not available on macOS CI") + } + db = testutil.SetupTestDB() + var err error + registry, err = NewNodeRegistry(db) + Expect(err).ToNot(HaveOccurred()) + }) + + seed := func(ctx context.Context, name string, totalDisk, availDisk uint64) string { + node := &BackendNode{ + Name: name, + NodeType: NodeTypeBackend, + Address: "10.0.0.1:50051", + TotalDisk: totalDisk, + AvailableDisk: availDisk, + } + Expect(registry.Register(ctx, node, true)).To(Succeed()) + return node.ID + } + + It("removes a full node from the candidate set", func(ctx SpecContext) { + fullID := seed(ctx, "nvidia-thor", 937*gb, 0) + roomyID := seed(ctx, "worker-roomy", 2000*gb, 1200*gb) + + got, err := registry.NarrowByDiskHeadroom(ctx, []string{fullID, roomyID}, DiskRequirementFor(int64(70*gb))) + + Expect(err).ToNot(HaveOccurred()) + Expect(got).To(ConsistOf(roomyID)) + }) + + It("fails with a capacity error naming the shortfall when no node can store the model", func(ctx SpecContext) { + fullID := seed(ctx, "nvidia-thor", 937*gb, 0) + + _, err := registry.NarrowByDiskHeadroom(ctx, []string{fullID}, DiskRequirementFor(int64(70*gb))) + + Expect(err).To(MatchError(ErrInsufficientDisk)) + // The operator must learn this at scheduling time, with the numbers + // that made the decision — not 16 minutes into a transfer. + Expect(err.Error()).To(ContainSubstring("nvidia-thor")) + }) + }) +}) + +var _ = Describe("scheduling a model onto a cluster without disk headroom", func() { + var ( + reg *fakeModelRouter + backend *holdBackend + unloader *fakeUnloader + router *SmartRouter + dir string + ) + + BeforeEach(func() { + reg = &fakeModelRouter{findAndLockErr: errors.New("not found")} + reg.findIdleNode = &BackendNode{ID: "n1", Name: "nvidia-thor", Address: "10.0.0.1:50051"} + backend = &holdBackend{} + unloader = &fakeUnloader{ + installReply: &messaging.BackendInstallReply{Success: true, Address: "10.0.0.1:9001"}, + } + router = NewSmartRouter(reg, SmartRouterOptions{ + Unloader: unloader, + ClientFactory: &holdClientFactory{client: backend}, + }) + dir = GinkgoT().TempDir() + }) + + route := func(modelFile string) error { + _, err := router.Route(context.Background(), "longcat-video-avatar-1.5", "models/big.gguf", "llama-cpp", + &pb.ModelOptions{Model: "models/big.gguf", ModelFile: modelFile}, false) + return err + } + + It("fails at scheduling time instead of installing a backend it cannot feed", func() { + reg.narrowByDiskErr = fmt.Errorf("%w: nvidia-thor has 0 B free of 937.0 GB", ErrInsufficientDisk) + + err := route(sparseCheckpoint(dir, "big.gguf", 70<<30)) + + Expect(err).To(MatchError(ErrInsufficientDisk)) + // The whole point is that nothing expensive happened: no backend + // install, so no staging, so no 17GB transferred over 16 minutes + // before the truth surfaced. + Expect(unloader.installCalls).To(BeEmpty()) + }) + + It("restores the pre-check behaviour when the operator disables the knob", func() { + // LOCALAI_DISTRIBUTED_DISK_HEADROOM_CHECK=false (or the runtime + // toggle) must give back exactly today's behaviour: the node that + // lacks space IS selected and the load proceeds. + reg.narrowByDiskErr = fmt.Errorf("%w: nvidia-thor has 0 B free of 937.0 GB", ErrInsufficientDisk) + router = NewSmartRouter(reg, SmartRouterOptions{ + Unloader: unloader, + ClientFactory: &holdClientFactory{client: backend}, + DiskHeadroomEnabled: func() bool { return false }, + }) + + Expect(route(sparseCheckpoint(dir, "big.gguf", 70<<30))).To(Succeed()) + Expect(unloader.installCalls).To(HaveLen(1)) + }) + + It("still evaluates the check when disabled, so the operator is not left blind", func() { + // Disabled means "do not block", NOT "do not look". A silently + // disabled safety check is how the original bug stayed invisible. + reg.narrowByDiskErr = fmt.Errorf("%w: nvidia-thor has 0 B free of 937.0 GB", ErrInsufficientDisk) + router = NewSmartRouter(reg, SmartRouterOptions{ + Unloader: unloader, + ClientFactory: &holdClientFactory{client: backend}, + DiskHeadroomEnabled: func() bool { return false }, + }) + + Expect(route(sparseCheckpoint(dir, "big.gguf", 70<<30))).To(Succeed()) + Expect(reg.narrowByDiskRequired).ToNot(BeEmpty(), + "the disabled check must still run so it can warn about the shortfall") + }) + + It("reads the toggle live, so a runtime change applies without a restart", func() { + // The router is constructed ONCE; the operator flips the setting + // afterwards. Snapshotting the value at construction would make the + // runtime setting a lie. + enabled := true + reg.narrowByDiskErr = fmt.Errorf("%w: nvidia-thor has 0 B free of 937.0 GB", ErrInsufficientDisk) + router = NewSmartRouter(reg, SmartRouterOptions{ + Unloader: unloader, + ClientFactory: &holdClientFactory{client: backend}, + DiskHeadroomEnabled: func() bool { return enabled }, + }) + + Expect(route(sparseCheckpoint(dir, "big.gguf", 70<<30))).To(MatchError(ErrInsufficientDisk)) + + enabled = false + Expect(route(sparseCheckpoint(dir, "big2.gguf", 70<<30))).To(Succeed()) + }) + + It("skips the check in shared-models mode, where nothing is staged to the worker", func() { + // With LOCALAI_DISTRIBUTED_SHARED_MODELS every node mounts the same + // models directory and stageModelFiles uploads nothing, so demanding + // 73GB free per node would reject a cluster that needs no new bytes + // at all. + reg.narrowByDiskErr = fmt.Errorf("%w: nvidia-thor has 0 B free of 937.0 GB", ErrInsufficientDisk) + router = NewSmartRouter(reg, SmartRouterOptions{ + Unloader: unloader, + ClientFactory: &holdClientFactory{client: backend}, + SharedModels: true, + }) + + Expect(route(sparseCheckpoint(dir, "big.gguf", 70<<30))).To(Succeed()) + Expect(reg.narrowByDiskRequired).To(BeEmpty(), + "shared-models mode stages nothing, so the check must not run at all") + }) + + It("sizes the disk requirement from the model, not from a fixed threshold", func() { + Expect(route(sparseCheckpoint(dir, "big.gguf", 70<<30))).To(Succeed()) + + Expect(reg.narrowByDiskRequired).ToNot(BeEmpty()) + // A 70 GiB checkpoint must ask for more than 70 GiB of free space. + Expect(reg.narrowByDiskRequired[0]).To(BeNumerically(">", uint64(70)<<30)) + }) +}) + +func nodeNames(list []BackendNode) []string { + out := make([]string, 0, len(list)) + for _, n := range list { + out = append(out, n.Name) + } + return out +} diff --git a/core/services/nodes/interfaces.go b/core/services/nodes/interfaces.go index 38aa34b49..399ad74f8 100644 --- a/core/services/nodes/interfaces.go +++ b/core/services/nodes/interfaces.go @@ -30,6 +30,7 @@ type ModelRouter interface { GetModelScheduling(ctx context.Context, modelName string) (*ModelSchedulingConfig, error) FindNodesBySelector(ctx context.Context, selector map[string]string) ([]BackendNode, error) FindNodesWithFreeSlot(ctx context.Context, modelName string, candidateNodeIDs []string) ([]BackendNode, error) + NarrowByDiskHeadroom(ctx context.Context, candidateNodeIDs []string, required uint64) ([]string, error) ReserveVRAM(ctx context.Context, nodeID string, bytes uint64) error ReleaseVRAM(ctx context.Context, nodeID string, bytes uint64) error FindNodeWithVRAMFromSet(ctx context.Context, minBytes uint64, nodeIDs []string) (*BackendNode, error) diff --git a/core/services/nodes/model_router_test.go b/core/services/nodes/model_router_test.go index 1670808c7..a0c14ab26 100644 --- a/core/services/nodes/model_router_test.go +++ b/core/services/nodes/model_router_test.go @@ -100,6 +100,9 @@ func (f *fakeModelRouterForSmartRouter) FindNodesBySelector(_ context.Context, _ func (f *fakeModelRouterForSmartRouter) FindNodesWithFreeSlot(_ context.Context, _ string, _ []string) ([]BackendNode, error) { return nil, nil } +func (f *fakeModelRouterForSmartRouter) NarrowByDiskHeadroom(_ context.Context, candidateNodeIDs []string, _ uint64) ([]string, error) { + return candidateNodeIDs, nil +} func (f *fakeModelRouterForSmartRouter) ReserveVRAM(_ context.Context, _ string, _ uint64) error { return nil } diff --git a/core/services/nodes/registry.go b/core/services/nodes/registry.go index 090f7008d..1f5c42ef7 100644 --- a/core/services/nodes/registry.go +++ b/core/services/nodes/registry.go @@ -35,9 +35,21 @@ type BackendNode struct { // reservation is only here to keep two scheduling decisions within the // same heartbeat window from over-committing the same node. ReservedVRAM uint64 `gorm:"column:reserved_vram;default:0" json:"reserved_vram"` - TotalRAM uint64 `gorm:"column:total_ram" json:"total_ram"` // Total system RAM in bytes (fallback when no GPU) - AvailableRAM uint64 `gorm:"column:available_ram" json:"available_ram"` // Available system RAM in bytes - GPUVendor string `gorm:"column:gpu_vendor;size:32" json:"gpu_vendor"` // nvidia, amd, intel, vulkan, unknown + TotalRAM uint64 `gorm:"column:total_ram" json:"total_ram"` // Total system RAM in bytes (fallback when no GPU) + AvailableRAM uint64 `gorm:"column:available_ram" json:"available_ram"` // Available system RAM in bytes + // TotalDisk / AvailableDisk describe the filesystem that BACKS THE WORKER'S + // MODELS DIRECTORY, not the root filesystem: staged weights are written + // there, so that is the only mount whose free space decides whether a + // staging request can succeed. Reported by the worker on registration and + // refreshed on every heartbeat. + // + // TotalDisk == 0 means "this worker does not report disk" (pre-upgrade + // worker, or a stat that failed) and is the ONLY value readers may treat as + // unknown. AvailableDisk == 0 is a real, actionable reading: it is exactly + // what a 100%-full node reports, and the case this pair exists to catch. + TotalDisk uint64 `gorm:"column:total_disk;default:0" json:"total_disk"` + AvailableDisk uint64 `gorm:"column:available_disk;default:0" json:"available_disk"` + GPUVendor string `gorm:"column:gpu_vendor;size:32" json:"gpu_vendor"` // nvidia, amd, intel, vulkan, unknown // GPUComputeCapability is the worker GPU's compute capability as // "major.minor" (e.g. "12.1" for GB10 / DGX Spark). Reported by the worker // on registration; used by the router to pick per-arch options (e.g. a @@ -94,6 +106,8 @@ const ( ColTotalVRAM = "total_vram" ColReservedVRAM = "reserved_vram" ColAvailableRAM = "available_ram" + ColTotalDisk = "total_disk" + ColAvailableDisk = "available_disk" ColGPUVendor = "gpu_vendor" ColGPUComputeCap = "gpu_compute_capability" ColLastHeartbeat = "last_heartbeat" @@ -422,6 +436,18 @@ func (r *NodeRegistry) Register(ctx context.Context, node *BackendNode, autoAppr return fmt.Errorf("clearing worker VRAM budget for node %s: %w", node.Name, err) } } + // Force-write the disk columns. Updates(struct) above zero-skips, and a + // worker whose models filesystem is 100% full re-registers with + // available_disk == 0 — the single most important reading there is. + // Zero-skipping it would leave the last healthy-looking value in place + // and put the full node straight back into rotation. + if err := r.db.WithContext(ctx).Model(&BackendNode{}).Where("id = ?", node.ID). + Updates(map[string]any{ + ColTotalDisk: node.TotalDisk, + ColAvailableDisk: node.AvailableDisk, + }).Error; err != nil { + return fmt.Errorf("recording disk capacity for node %s: %w", node.Name, err) + } // Preserve auth references from existing record. // GORM Updates(struct) skips zero-value fields, so the DB retains // the old auth_user_id/api_key_id but the caller's struct is empty. @@ -687,6 +713,11 @@ type HeartbeatUpdate struct { AvailableVRAM *uint64 `json:"available_vram,omitempty"` TotalVRAM *uint64 `json:"total_vram,omitempty"` AvailableRAM *uint64 `json:"available_ram,omitempty"` + // AvailableDisk / TotalDisk describe the worker's models filesystem. + // Pointers so a worker that cannot read them omits the fields rather than + // reporting a zero the scheduler would act on. + AvailableDisk *uint64 `json:"available_disk,omitempty"` + TotalDisk *uint64 `json:"total_disk,omitempty"` GPUVendor string `json:"gpu_vendor,omitempty"` } @@ -720,6 +751,14 @@ func (r *NodeRegistry) Heartbeat(ctx context.Context, nodeID string, update *Hea if update.AvailableRAM != nil { updates[ColAvailableRAM] = *update.AvailableRAM } + // Written unconditionally when reported, INCLUDING zero: a full disk + // reports 0 free, and that is the reading the scheduler must act on. + if update.AvailableDisk != nil { + updates[ColAvailableDisk] = *update.AvailableDisk + } + if update.TotalDisk != nil { + updates[ColTotalDisk] = *update.TotalDisk + } if update.GPUVendor != "" { updates[ColGPUVendor] = update.GPUVendor } diff --git a/core/services/nodes/router.go b/core/services/nodes/router.go index a463ef07a..5b6aed953 100644 --- a/core/services/nodes/router.go +++ b/core/services/nodes/router.go @@ -66,6 +66,11 @@ type SmartRouterOptions struct { // The reconciler reads the same instance to autoscale a saturated cache-warm // replica. nil disables recording (the disabled path stays a no-op). Pressure *prefixcache.Pressure + // DiskHeadroomEnabled is read LIVE on every scheduling decision (not + // snapshotted at construction) so the operator's runtime toggle takes + // effect without a restart. nil means enabled — the safe default, and what + // every embedder/test that never wires the knob gets. + DiskHeadroomEnabled func() bool // SharedModels asserts that every node mounts the same models directory at // the same path. When true, stageModelFiles skips all uploading and leaves // the absolute model paths untouched so the worker loads them directly from @@ -167,6 +172,10 @@ type SmartRouter struct { // sharedModels skips file staging when all nodes mount the same models // directory at the same path (see SmartRouterOptions.SharedModels). sharedModels bool + // diskHeadroomEnabled is the live read of the operator's disk-headroom + // toggle (see SmartRouterOptions.DiskHeadroomEnabled). Never nil after + // NewSmartRouter. + diskHeadroomEnabled func() bool // modelLoadCeiling bounds how long a cold load may hold the per-model // advisory lock (see SmartRouterOptions.ModelLoadCeiling). modelLoadCeiling time.Duration @@ -202,22 +211,28 @@ func NewSmartRouter(registry ModelRouter, opts SmartRouterOptions) *SmartRouter if ceiling <= 0 { ceiling = ModelLoadCeilingFor(config.DefaultBackendInstallTimeout, loadTimeout) } + // Default ON: a nil provider must not silently disable a safety check. + diskHeadroom := opts.DiskHeadroomEnabled + if diskHeadroom == nil { + diskHeadroom = func() bool { return true } + } return &SmartRouter{ - registry: registry, - unloader: opts.Unloader, - fileStager: opts.FileStager, - galleriesJSON: opts.GalleriesJSON, - clientFactory: factory, - db: opts.DB, - stagingTracker: NewStagingTracker(), - conflictResolver: opts.ConflictResolver, - probeCache: newProbeCache(probeCacheTTL), - prefixProvider: opts.PrefixProvider, - prefixConfig: opts.PrefixConfig, - pressure: opts.Pressure, - sharedModels: opts.SharedModels, - modelLoadCeiling: ceiling, - modelLoadTimeout: loadTimeout, + registry: registry, + unloader: opts.Unloader, + fileStager: opts.FileStager, + galleriesJSON: opts.GalleriesJSON, + clientFactory: factory, + db: opts.DB, + stagingTracker: NewStagingTracker(), + conflictResolver: opts.ConflictResolver, + probeCache: newProbeCache(probeCacheTTL), + prefixProvider: opts.PrefixProvider, + prefixConfig: opts.PrefixConfig, + pressure: opts.Pressure, + sharedModels: opts.SharedModels, + diskHeadroomEnabled: diskHeadroom, + modelLoadCeiling: ceiling, + modelLoadTimeout: loadTimeout, // Zero values are resolved to their defaults inside // newLoadDeadlineContext, which also clamps the stall window to the // ceiling, so nothing to normalize here. @@ -941,6 +956,18 @@ func (r *SmartRouter) scheduleNewModel(ctx context.Context, backendType, modelID return nil, "", 0, err } + // Narrow candidates to nodes that can physically STORE the model. + // + // Staging writes the checkpoint into the worker's models directory before + // the backend ever sees it, so a node without room there cannot succeed no + // matter how much VRAM it advertises. Doing this here, before the replica + // slot and VRAM passes, is the whole point: the alternative is discovering + // it sixteen minutes into a transfer, from the worker, as a 500. + candidateNodeIDs, err = r.narrowByDiskHeadroom(ctx, modelID, modelOpts, candidateNodeIDs) + if err != nil { + return nil, "", 0, err + } + // Narrow candidates to nodes that still have a free replica slot for this // model. Without this filter, the scheduler would happily pick a node // already at capacity for this model (e.g. when MinReplicas > free @@ -1056,6 +1083,54 @@ func (r *SmartRouter) scheduleNewModel(ctx context.Context, backendType, modelID return node, addr, replicaIdx, nil } +// narrowByDiskHeadroom drops candidate nodes that cannot store the model and +// returns the surviving set. +// +// When nothing fits it returns an error wrapping ErrInsufficientDisk so the +// caller fails at scheduling time — unless the operator disabled the check, in +// which case it warns and hands back the original candidates unchanged. +// +// modelPayloadBytes stats the same local paths stageModelFiles uploads, which +// is why the requirement can be answered before a node is chosen at all. +func (r *SmartRouter) narrowByDiskHeadroom(ctx context.Context, modelID string, modelOpts *pb.ModelOptions, candidateNodeIDs []string) ([]string, error) { + // Shared-models mode uploads nothing: every node already mounts this exact + // models directory at this exact path (stageModelFiles returns early). + // Demanding the full checkpoint size of free space per node would reject a + // cluster that needs no new bytes at all. + if r.sharedModels { + return candidateNodeIDs, nil + } + + requiredDisk := DiskRequirementFor(modelPayloadBytes(modelOpts)) + diskCandidates, diskErr := r.registry.NarrowByDiskHeadroom(ctx, candidateNodeIDs, requiredDisk) + + // The check runs even when disabled. "Disabled" means do not BLOCK, not do + // not LOOK: a safety check that goes quiet when switched off is how the + // original incident stayed invisible for sixteen minutes. The operator + // surrenders the veto and keeps the diagnosis. + if !r.diskHeadroomEnabled() { + if errors.Is(diskErr, ErrInsufficientDisk) { + xlog.Warn("No node has room to store this model, but the disk-headroom check is DISABLED; scheduling anyway — staging will most likely fail with ENOSPC", + "model", modelID, "knob", config.FlagDiskHeadroomCheck, "detail", diskErr) + } + return candidateNodeIDs, nil + } + + switch { + case errors.Is(diskErr, ErrInsufficientDisk): + // Fail here rather than picking a node and letting staging discover it. + return nil, fmt.Errorf("scheduling %s: %w", modelID, diskErr) + case diskErr != nil: + // A registry read failure is not a capacity verdict. Log and schedule + // with the unnarrowed set — the old (worse) behaviour, but never a + // wedged cluster because a query hiccuped. + xlog.Warn("Failed to check node disk headroom; scheduling without the disk filter", + "model", modelID, "required", vram.FormatBytes(requiredDisk), "error", diskErr) + return candidateNodeIDs, nil + } + return diskCandidates, nil +} + // estimateModelVRAM estimates the VRAM required for a model using the unified estimator. func (r *SmartRouter) estimateModelVRAM(ctx context.Context, opts *pb.ModelOptions) uint64 { estCtx, cancel := context.WithTimeout(ctx, 10*time.Second) diff --git a/core/services/nodes/router_test.go b/core/services/nodes/router_test.go index 81696ac4d..8f0b2f0fd 100644 --- a/core/services/nodes/router_test.go +++ b/core/services/nodes/router_test.go @@ -117,6 +117,13 @@ type fakeModelRouter struct { loadedReplicaStatsByName map[string][]ReplicaCandidate loadedReplicaStatsErr error + // NarrowByDiskHeadroom returns. Default (zero value) passes the candidate + // set through untouched, matching a cluster with ample free disk. + narrowByDiskIDs []string + narrowByDiskErr error + // Free bytes demanded by each NarrowByDiskHeadroom call, in call order. + narrowByDiskRequired []uint64 + // Track calls for assertions decrementCalls []string // "nodeID:modelName" incrementCalls []string @@ -231,6 +238,17 @@ func (f *fakeModelRouter) FindNodesWithFreeSlot(_ context.Context, _ string, _ [ return f.findBySelectorNodes, f.findBySelectorErr } +func (f *fakeModelRouter) NarrowByDiskHeadroom(_ context.Context, candidateNodeIDs []string, required uint64) ([]string, error) { + f.narrowByDiskRequired = append(f.narrowByDiskRequired, required) + if f.narrowByDiskErr != nil { + return nil, f.narrowByDiskErr + } + if f.narrowByDiskIDs != nil { + return f.narrowByDiskIDs, nil + } + return candidateNodeIDs, nil +} + func (f *fakeModelRouter) ReserveVRAM(_ context.Context, _ string, _ uint64) error { return nil } diff --git a/core/services/worker/registration.go b/core/services/worker/registration.go index 6b4ce19d6..186bcd1b0 100644 --- a/core/services/worker/registration.go +++ b/core/services/worker/registration.go @@ -157,6 +157,20 @@ func (cfg *Config) registrationBody() map[string]any { "max_replicas_per_model": maxReplicas, } + // Report free space on the filesystem that backs the MODELS directory. + // That is where staged weights land, so it is the only mount whose free + // space decides whether this node can accept a model — the scheduler uses + // it to avoid picking a node that would fail with ENOSPC mid-transfer. + if diskInfo, err := xsysinfo.GetDiskInfo(cfg.ModelsPath); err != nil { + // Omitted, not zeroed: total_disk == 0 is how the frontend recognises + // "this worker does not report disk" and keeps it in rotation. + xlog.Warn("Failed to detect worker models-path disk capacity; registering without it", + "path", cfg.ModelsPath, "error", err) + } else { + body["total_disk"] = diskInfo.Total + body["available_disk"] = diskInfo.Available + } + // Report the operator-set budget as a STRING so the server resolves and // enforces it against the raw VRAM above. The worker never caps its own // total_vram/available_vram, and never touches the xsysinfo process-global @@ -221,5 +235,17 @@ func (cfg *Config) heartbeatBody() map[string]any { body["available_ram"] = ramInfo.Available } } + + // Free disk changes far faster than VRAM under staging traffic (each + // accepted model permanently consumes space), so it has to be refreshed + // every heartbeat rather than only at registration — a node that filled up + // hours after registering is precisely the case that broke. + if diskInfo, err := xsysinfo.GetDiskInfo(cfg.ModelsPath); err != nil { + xlog.Debug("Failed to detect worker models-path disk capacity for heartbeat", + "path", cfg.ModelsPath, "error", err) + } else { + body["total_disk"] = diskInfo.Total + body["available_disk"] = diskInfo.Available + } return body } diff --git a/docs/content/features/distributed-mode.md b/docs/content/features/distributed-mode.md index 5db0b9f21..f2c45e02e 100644 --- a/docs/content/features/distributed-mode.md +++ b/docs/content/features/distributed-mode.md @@ -25,12 +25,13 @@ Distributed mode requires authentication enabled with a **PostgreSQL** database The SmartRouter uses **idle-first** scheduling with **preemptive eviction**: 1. If the model is already loaded on a node → use it (per-model gRPC address) -2. If no node has the model → prefer nodes with enough free VRAM -3. Fall back to idle nodes (zero models), then least-loaded nodes -4. If no node has capacity → **evict the least-recently-used model with zero in-flight requests** to free a node -5. If all models are busy → wait (with timeout) for a model to become idle, then evict -6. Send `backend.install` NATS event with backend name + model ID → worker starts a new gRPC process on a dynamic port -7. SmartRouter calls gRPC `LoadModel` on the model-specific port, records in DB +2. Drop any node without room to **store** the model on its models filesystem (see [Disk headroom](#disk-headroom)) +3. If no node has the model → prefer nodes with enough free VRAM +4. Fall back to idle nodes (zero models), then least-loaded nodes +5. If no node has capacity → **evict the least-recently-used model with zero in-flight requests** to free a node +6. If all models are busy → wait (with timeout) for a model to become idle, then evict +7. Send `backend.install` NATS event with backend name + model ID → worker starts a new gRPC process on a dynamic port +8. SmartRouter calls gRPC `LoadModel` on the model-specific port, records in DB Each model gets its own gRPC backend process, so a single worker can serve multiple models simultaneously (e.g., a chat model and an embedding model). @@ -68,6 +69,7 @@ The frontend is a standard LocalAI instance with distributed mode enabled. These | `--distributed-require-auth` | `LOCALAI_DISTRIBUTED_REQUIRE_AUTH` | `false` | **Umbrella switch.** Implies both `--nats-require-auth` and `--registration-require-auth` - one knob to lock down the NATS bus *and* the registration/file-transfer layer. Set this in production instead of the two granular flags. | | `--auto-approve-nodes` | `LOCALAI_AUTO_APPROVE_NODES` | `false` | Auto-approve new worker nodes (skip admin approval) | | `--distributed-shared-models` | `LOCALAI_DISTRIBUTED_SHARED_MODELS` | `false` | Assert that every node mounts the **same** models directory at the **same** path (a shared volume). When `true`, the router skips file staging entirely and workers load models directly from the shared path instead of re-downloading them. See [Shared models directory](#shared-models-directory). | +| `--distributed-disk-headroom-check` | `LOCALAI_DISTRIBUTED_DISK_HEADROOM_CHECK` | `true` | Reject worker nodes that lack free space to store the model, at scheduling time rather than partway through staging. When `false`, node selection ignores free disk; the check still runs and warns when it would have rejected every node. Also toggleable at runtime via the `distributed_disk_headroom_check` setting. See [Disk headroom](#disk-headroom). | | `--auth` | `LOCALAI_AUTH` | `false` | **Must be `true`** for distributed mode | | `--auth-database-url` | `LOCALAI_AUTH_DATABASE_URL` | *(required)* | PostgreSQL connection URL | | `--backend-install-timeout` | `LOCALAI_NATS_BACKEND_INSTALL_TIMEOUT` | `15m` | How long the frontend waits for a worker to acknowledge a backend install before considering the request stalled. Raise it when workers pull large backend images over slow links. If a worker takes longer than this, the operation shows as "still installing in background" in the admin UI and clears once the worker finishes. | @@ -457,6 +459,39 @@ curl -X DELETE http://frontend:8080/api/nodes//vram-budget \ The value accepts the same formats as the standalone budget: a percentage (`80%`) or an absolute amount (`12GB`, `12GiB`, `12000MB`, or raw bytes). It is a **hard ceiling**: the node's advertised VRAM becomes `min(detected, budget)`, so a budget can only lower the number, never raise it above the hardware. An admin-set node budget is **sticky across worker restarts**: it is stored in the node registry and reapplied when the worker re-registers, so it wins over whatever the worker reports on reconnect. For the underlying semantics and the standalone equivalent, see [VRAM Budget]({{%relref "advanced/vram-management#vram-budget-allocation-ceiling" %}}). +### Disk headroom + +Model weights are **staged onto the worker's disk** before the backend loads them, so a node needs free space as well as free VRAM. Each worker reports the capacity of the filesystem backing its **models directory** (`--models-path`), not the root filesystem, on registration and on every heartbeat. Those figures appear as `total_disk` and `available_disk` in the nodes API and as **Models disk free** on the node detail page. + +Before placing a model, the SmartRouter removes any node whose models filesystem cannot hold it. The requirement is derived from the model's actual on-disk size plus a small margin (5%, at least 1 GiB), rather than a fixed percentage of the node's disk — a fixed threshold would take a small-but-usable node out of rotation for models it could comfortably store. When the model's size cannot be determined locally (a bare HuggingFace repo id that the worker fetches itself), the node only has to clear a 2 GiB floor. + +If **no** node has enough space, the request fails immediately with a capacity error naming the requirement and each node's free space, for example: + +``` +scheduling longcat-video-avatar-1.5: no node has enough free disk for the model: +need 73.5 GB free on the models filesystem, but nvidia-thor has 0 B free of 937.0 GB +``` + +This is deliberately a scheduling-time verdict. Without it, a worker with a full disk still reported `status: healthy`, accepted the staging request, transferred tens of gigabytes and only then failed with `no space left on device` — minutes after a decision that could never have succeeded. + +Workers that predate this feature (or whose disk reading fails) report `total_disk` as `0`. Such nodes are treated as *unknown*, not *full*, and stay in rotation, so a rolling upgrade never empties the candidate pool. A full disk is distinguishable because it reports a non-zero `total_disk` with `available_disk` at `0`. + +Low disk does **not** mark a node `unhealthy`. Disk is compared per model rather than against a global threshold, so a node that is too small for one model remains a valid target for smaller ones. The check is also skipped entirely in [shared-models mode](#shared-models-directory), where nothing is staged to the worker at all. + +#### Turning the check off + +The check is **on by default**. To disable it, start the frontend with `--distributed-disk-headroom-check=false` / `LOCALAI_DISTRIBUTED_DISK_HEADROOM_CHECK=false`, or toggle **Settings → Distributed → Disk headroom check** in the WebUI (`distributed_disk_headroom_check` via `POST /api/settings`). The runtime setting takes effect on the next placement, with no restart; the env/CLI flag only sets the value LocalAI boots with, and both write the same underlying value, so the last change wins. + +Disabling means **warn, do not block**. Node selection goes back to ignoring free disk (the pre-check behaviour), but the check still runs, and when it would have rejected *every* node it logs a warning naming the shortfall: + +``` +WARN No node has room to store this model, but the disk-headroom check is DISABLED; + scheduling anyway — staging will most likely fail with ENOSPC + model=longcat-video-avatar-1.5 knob=distributed-disk-headroom-check +``` + +The alternative — skipping the check outright — was rejected because it reproduces the condition that made the original bug expensive: the cluster was doing something that could not work and said nothing about it. The escape hatch exists for setups where the size estimate is wrong (deduplicating or compressing filesystems, a backend that fetches its own weights rather than using the staged copy), and in exactly those cases the operator needs to see what LocalAI thought was wrong. Disabling is logged once at startup as well. + The **LocalAI Assistant** can also set a node budget conversationally through the `set_node_vram_budget` MCP tool. ## Node Approval diff --git a/pkg/xsysinfo/disk.go b/pkg/xsysinfo/disk.go new file mode 100644 index 000000000..c531808b4 --- /dev/null +++ b/pkg/xsysinfo/disk.go @@ -0,0 +1,68 @@ +package xsysinfo + +import ( + "fmt" + "os" + "path/filepath" + + "github.com/shirou/gopsutil/v3/disk" +) + +// DiskInfo describes the filesystem backing a particular path. +type DiskInfo struct { + Total uint64 `json:"total"` + Used uint64 `json:"used"` + Available uint64 `json:"available"` +} + +// GetDiskInfo reports usage for the filesystem that holds `path`. +// +// The path matters. A worker stages model weights into its models directory, +// which is very often a separate mount (an NVMe volume, a network share) from +// `/`. Measuring the root filesystem would answer a question nobody asked and +// would have missed the incident this exists for, where the models volume was +// the one at 100%. +// +// A path that does not exist yet is normal on a fresh worker: the models +// directory is created lazily. Rather than failing, walk up to the nearest +// existing ancestor, which sits on the same filesystem the directory will be +// created on in all but pathological setups. +func GetDiskInfo(path string) (*DiskInfo, error) { + if path == "" { + return nil, fmt.Errorf("no path given") + } + target, err := nearestExistingDir(path) + if err != nil { + return nil, err + } + usage, err := disk.Usage(target) + if err != nil { + return nil, fmt.Errorf("reading disk usage for %s: %w", target, err) + } + return &DiskInfo{ + Total: usage.Total, + Used: usage.Used, + // Free, not (Total - Used): on ext4 the reserved-blocks pool sits + // between them, and writing into it is exactly what fails with + // ENOSPC for a non-root process. + Available: usage.Free, + }, nil +} + +// nearestExistingDir walks up from path until it finds something that exists. +func nearestExistingDir(path string) (string, error) { + abs, err := filepath.Abs(path) + if err != nil { + return "", fmt.Errorf("resolving %s: %w", path, err) + } + for { + if _, err := os.Stat(abs); err == nil { + return abs, nil + } + parent := filepath.Dir(abs) + if parent == abs { + return "", fmt.Errorf("no existing ancestor of %s", path) + } + abs = parent + } +} diff --git a/pkg/xsysinfo/disk_test.go b/pkg/xsysinfo/disk_test.go new file mode 100644 index 000000000..ebd9ae1c4 --- /dev/null +++ b/pkg/xsysinfo/disk_test.go @@ -0,0 +1,36 @@ +package xsysinfo_test + +import ( + "path/filepath" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + + "github.com/mudler/LocalAI/pkg/xsysinfo" +) + +var _ = Describe("GetDiskInfo", func() { + It("reports the filesystem holding an existing directory", func() { + info, err := xsysinfo.GetDiskInfo(GinkgoT().TempDir()) + + Expect(err).ToNot(HaveOccurred()) + Expect(info.Total).To(BeNumerically(">", 0)) + Expect(info.Available).To(BeNumerically("<=", info.Total)) + }) + + It("measures the nearest existing ancestor when the models dir is not created yet", func() { + // A fresh worker registers before its models directory exists; the + // capacity report must still be a real reading rather than an error. + notYet := filepath.Join(GinkgoT().TempDir(), "models", "subdir") + + info, err := xsysinfo.GetDiskInfo(notYet) + + Expect(err).ToNot(HaveOccurred()) + Expect(info.Total).To(BeNumerically(">", 0)) + }) + + It("errors rather than guessing when given no path", func() { + _, err := xsysinfo.GetDiskInfo("") + Expect(err).To(HaveOccurred()) + }) +})