mirror of
https://github.com/mudler/LocalAI.git
synced 2026-07-30 18:09:05 -04:00
6584db992f
* 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 <mudler@localai.io>
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 <mudler@localai.io>
Assisted-by: Claude Code:claude-opus-4-8[1m] [Read] [Edit] [Bash]
---------
Signed-off-by: Ettore Di Giacinto <mudler@localai.io>
Co-authored-by: Ettore Di Giacinto <mudler@localai.io>
122 lines
6.6 KiB
Go
122 lines
6.6 KiB
Go
package nodes
|
|
|
|
import (
|
|
"context"
|
|
"time"
|
|
|
|
grpc "github.com/mudler/LocalAI/pkg/grpc"
|
|
)
|
|
|
|
// ModelRouter is used by SmartRouter for routing decisions and model lifecycle.
|
|
type ModelRouter interface {
|
|
FindAndLockNodeWithModel(ctx context.Context, modelName string, candidateNodeIDs []string, pref *RoutePreference) (*BackendNode, *NodeModel, error)
|
|
DecrementInFlight(ctx context.Context, nodeID, modelName string, replicaIndex int) error
|
|
IncrementInFlight(ctx context.Context, nodeID, modelName string, replicaIndex int) error
|
|
RemoveNodeModel(ctx context.Context, nodeID, modelName string, replicaIndex int) error
|
|
RemoveAllNodeModelReplicas(ctx context.Context, nodeID, modelName string) error
|
|
TouchNodeModel(ctx context.Context, nodeID, modelName string, replicaIndex int)
|
|
SetNodeModel(ctx context.Context, nodeID, modelName string, replicaIndex int, state, address string, initialInFlight int) error
|
|
SetNodeModelLoadInfo(ctx context.Context, nodeID, modelName string, replicaIndex int, backendType string, optsBlob []byte) error
|
|
UpsertModelLoadInfo(ctx context.Context, modelName, backendType string, optsBlob []byte) error
|
|
GetModelLoadInfo(ctx context.Context, modelName string) (backendType string, optsBlob []byte, err error)
|
|
NextFreeReplicaIndex(ctx context.Context, nodeID, modelName string, maxSlots int) (int, error)
|
|
CountReplicasOnNode(ctx context.Context, nodeID, modelName string) (int, error)
|
|
FindNodeWithVRAM(ctx context.Context, minBytes uint64) (*BackendNode, error)
|
|
FindIdleNode(ctx context.Context) (*BackendNode, error)
|
|
FindLeastLoadedNode(ctx context.Context) (*BackendNode, error)
|
|
FindGlobalLRUModelWithZeroInFlight(ctx context.Context) (*NodeModel, error)
|
|
FindLRUModel(ctx context.Context, nodeID string) (*NodeModel, error)
|
|
Get(ctx context.Context, nodeID string) (*BackendNode, error)
|
|
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)
|
|
FindIdleNodeFromSet(ctx context.Context, nodeIDs []string) (*BackendNode, error)
|
|
FindLeastLoadedNodeFromSet(ctx context.Context, nodeIDs []string) (*BackendNode, error)
|
|
GetNodeLabels(ctx context.Context, nodeID string) ([]NodeLabel, error)
|
|
FindNodesWithModel(ctx context.Context, modelName string) ([]BackendNode, error)
|
|
LoadedReplicaStats(ctx context.Context, modelName string, candidateNodeIDs []string) ([]ReplicaCandidate, error)
|
|
}
|
|
|
|
// ConcurrencyConflictResolver returns the names of configured models that
|
|
// share at least one concurrency group with the given model. It is satisfied
|
|
// by *config.ModelConfigLoader and lets the SmartRouter make group-aware
|
|
// placement decisions without importing the config package's full surface.
|
|
type ConcurrencyConflictResolver interface {
|
|
GetModelsConflictingWith(modelName string) []string
|
|
}
|
|
|
|
// NodeHealthStore is used by HealthMonitor for node status management.
|
|
type NodeHealthStore interface {
|
|
List(ctx context.Context) ([]BackendNode, error)
|
|
GetNodeModels(ctx context.Context, nodeID string) ([]NodeModel, error)
|
|
MarkOffline(ctx context.Context, nodeID string) error
|
|
MarkUnhealthy(ctx context.Context, nodeID string) error
|
|
MarkHealthy(ctx context.Context, nodeID string) error
|
|
Heartbeat(ctx context.Context, nodeID string, update *HeartbeatUpdate) error
|
|
FindStaleNodes(ctx context.Context, threshold time.Duration) ([]BackendNode, error)
|
|
RemoveNodeModel(ctx context.Context, nodeID, modelName string, replicaIndex int) error
|
|
}
|
|
|
|
// ModelLocator is used by RemoteUnloaderAdapter for model discovery.
|
|
type ModelLocator interface {
|
|
FindNodesWithModel(ctx context.Context, modelName string) ([]BackendNode, error)
|
|
RemoveNodeModel(ctx context.Context, nodeID, modelName string, replicaIndex int) error
|
|
RemoveAllNodeModelReplicas(ctx context.Context, nodeID, modelName string) error
|
|
}
|
|
|
|
// ModelLookup is used by DistributedModelStore for model existence queries.
|
|
type ModelLookup interface {
|
|
FindNodeForModel(ctx context.Context, modelName string) (*BackendNode, bool)
|
|
ListAllLoadedModels(ctx context.Context) ([]NodeModel, error)
|
|
Get(ctx context.Context, nodeID string) (*BackendNode, error)
|
|
}
|
|
|
|
// InFlightTracker is used by InFlightTrackingClient for request counting.
|
|
type InFlightTracker interface {
|
|
IncrementInFlight(ctx context.Context, nodeID, modelName string, replicaIndex int) error
|
|
DecrementInFlight(ctx context.Context, nodeID, modelName string, replicaIndex int) error
|
|
// RemoveNodeModel drops a stale replica row so the next request reloads the
|
|
// model instead of routing back to a node where it is no longer loaded.
|
|
RemoveNodeModel(ctx context.Context, nodeID, modelName string, replicaIndex int) error
|
|
}
|
|
|
|
// NodeManager is used by HTTP endpoints for node registration and lifecycle.
|
|
type NodeManager interface {
|
|
Register(ctx context.Context, node *BackendNode, autoApprove bool) error
|
|
Get(ctx context.Context, nodeID string) (*BackendNode, error)
|
|
GetByName(ctx context.Context, name string) (*BackendNode, error)
|
|
List(ctx context.Context) ([]BackendNode, error)
|
|
Deregister(ctx context.Context, nodeID string) error
|
|
ApproveNode(ctx context.Context, nodeID string) error
|
|
MarkOffline(ctx context.Context, nodeID string) error
|
|
MarkDraining(ctx context.Context, nodeID string) error
|
|
MarkHealthy(ctx context.Context, nodeID string) error
|
|
Heartbeat(ctx context.Context, nodeID string, update *HeartbeatUpdate) error
|
|
GetNodeModels(ctx context.Context, nodeID string) ([]NodeModel, error)
|
|
UpdateAuthRefs(ctx context.Context, nodeID, authUserID, apiKeyID string) error
|
|
RemoveNodeModel(ctx context.Context, nodeID, modelName string, replicaIndex int) error
|
|
RemoveAllNodeModelReplicas(ctx context.Context, nodeID, modelName string) error
|
|
}
|
|
|
|
// BackendClientFactory creates gRPC backend clients.
|
|
type BackendClientFactory interface {
|
|
NewClient(address string, parallel bool) grpc.Backend
|
|
}
|
|
|
|
// tokenClientFactory is the default BackendClientFactory that creates gRPC
|
|
// clients with an optional bearer token for distributed auth.
|
|
type tokenClientFactory struct {
|
|
token string
|
|
}
|
|
|
|
func (f *tokenClientFactory) NewClient(address string, parallel bool) grpc.Backend {
|
|
if f.token != "" {
|
|
return grpc.NewClientWithToken(address, parallel, nil, false, f.token)
|
|
}
|
|
return grpc.NewClient(address, parallel, nil, false)
|
|
}
|