mirror of
https://github.com/mudler/LocalAI.git
synced 2026-10-05 12:34:43 -04:00
* feat(messaging): add shared subject rules Assisted-by: Claude:claude-sonnet-5-5 Signed-off-by: Ettore Di Giacinto <mudler@localai.io> * test(messaging): cover BroadcastRoots, ControlRoots and SubjectRoot Assisted-by: Claude:claude-sonnet-5-5 Signed-off-by: Ettore Di Giacinto <mudler@localai.io> * feat(messaging): add Broadcaster and enforce subject rules in every carrier Broadcaster is the fan-out half of MessagingClient. The NATS client and the in-memory FakeBus now refuse a subject outside the served roots and any wildcard other than a whole single token, and FakeBus shares MatchSubject instead of its own copy. FakeBus Unsubscribe now removes its own subscription instead of the first one with the same subject. A shared conformance suite in messagingtest runs against both carriers. The distributed e2e specs that used invented test.* subjects, and the one that subscribed with a > filter, now use subjects from subjects.go. Assisted-by: Claude:claude-sonnet-5-5 Signed-off-by: Ettore Di Giacinto <mudler@localai.io> * refactor: depend on Broadcaster where only publish and subscribe are used Narrowed to messaging.Broadcaster: nodes/staging_progress.go, nodes/install_progress_publisher.go, galleryop/operation.go, galleryop/service.go, agentpool/user_services.go, agentpool/agent_jobs.go, openresponses/store.go, openresponses/sync.go, syncstate/syncstate.go, finetune/service.go, quantization/service.go and failover/distsync/distsync.go. SubscribeJSON now takes a Broadcaster because it only calls Subscribe, which lets the narrowed consumers use it. Stayed wide: worker/supervisor.go, because its client field also serves the SubscribeReply handlers in worker/lifecycle.go. The request/reply, queue and wiring files (nodes/unloader.go, nodes/file_stager_s3.go, jobs/dispatcher.go, agents/dispatcher.go, agents/events.go, worker/file_staging.go, cli/agent_worker.go, http/app.go) are unchanged by design. Assisted-by: Claude:claude-sonnet-5-5 Signed-off-by: Ettore Di Giacinto <mudler@localai.io> * refactor(nodes): name the no-route condition and confine the carrier error Consumers matched nats.ErrNoResponders, which names an absence, to demote a node. They now match ErrNoRoute, the control path maps the carrier's failure onto it, and timeouts and worker refusals are pinned as not being no-route. Assisted-by: Claude:claude-sonnet-5-5 Signed-off-by: Ettore Di Giacinto <mudler@localai.io> * docs(nodes): state which FileStager implementations return ErrNoRoute Assisted-by: Claude:claude-sonnet-5-5 Signed-off-by: Ettore Di Giacinto <mudler@localai.io> * refactor(nodes): build backend clients through one node-aware seam Assisted-by: Claude:claude-sonnet-5-5 Signed-off-by: Ettore Di Giacinto <mudler@localai.io> * docs: describe the distributed transport seams Assisted-by: Claude:claude-sonnet-5-5 Signed-off-by: Ettore Di Giacinto <mudler@localai.io> * docs: correct comments that overclaim after the seams refactor Assisted-by: Claude:claude-sonnet-5-5 Signed-off-by: Ettore Di Giacinto <mudler@localai.io> * fix(agent-worker): refuse an unserved LOCALAI_AGENT_SUBJECT at startup The messaging client now refuses a subject whose root no carrier serves. An agent worker started with a custom LOCALAI_AGENT_SUBJECT such as tenant-a.agent.execute used to start and then wait on a subject the frontend never publishes to. After the subject rules landed it exited at subscribe time with an error that did not name the setting. Behaviour change: the worker now checks LOCALAI_AGENT_SUBJECT before it registers or connects, and exits with an error that names the variable and says to use a served subject under the agent root, for example agent.execute. The served roots are not widened: a custom root was never delivered by the frontend, and a wider set would reopen the drift the subject rules exist to close. The flag help and the agent worker docs state the constraint. Assisted-by: Claude:claude-sonnet-5-5 Signed-off-by: Ettore Di Giacinto <mudler@localai.io> * test(nodes): pin the reactions to ErrNoRoute Three callers react to ErrNoRoute and had no spec: the reconciler's upgrade drain falls back to the legacy forced install, the reconciler marks the node unhealthy when a pending op has no route, and the backend-op fan-out marks the node unhealthy. Each spec drives the real caller with a scripted no-responders reply and reads the result from the registry or the recorded requests. A fourth spec pins the other side: a pending op that times out leaves the node healthy and only counts the attempt, so mapping timeouts onto ErrNoRoute would fail here. Assisted-by: Claude:claude-sonnet-5-5 Signed-off-by: Ettore Di Giacinto <mudler@localai.io> * test(messaging): pin client subject checks and fail the carrier suite in CI Add specs that call Publish, Request, Subscribe, QueueSubscribe, SubscribeReply and QueueSubscribeReply on a client with no connection. Each call must return ErrUnservedSubject for bogus.thing and ErrUnsupportedWildcard for jobs.>. This proves that the subject check runs before the connection is used, and needs no server. The NATS conformance suite is the only check that runs the subject rules against a real carrier. Before this change it skipped without output when Docker was missing. Now it fails when CI is set, so a Linux runner without Docker cannot hide it. It still skips on local runs and on macOS CI, which has no Docker. Add SubjectNodeBackendInstallProgress to the list of constructors that must build served subjects, and ask contributors to extend the list. Assisted-by: Claude:claude-sonnet-5-5 Signed-off-by: Ettore Di Giacinto <mudler@localai.io> * docs: state what ErrNoRoute may change, and group the distributed guides The seams note said MarkUnhealthy was the only state change allowed on ErrNoRoute. A pending backend op still records the failed attempt, counts toward the reconciler's retry limit and is dead-lettered after the maximum attempts. The note now says that MarkUnhealthy is the only change to the node's own state, and that the per-op accounting is not a verdict about the node. The note also documents that the NATS conformance run fails under CI when Docker is missing. The distributed-seams row moves next to the distributed-state row in the topics table. The liveness ping spec header now says no route is a reason to skip the worker, not proof that the worker is gone. Assisted-by: Claude:claude-sonnet-5-5 Signed-off-by: Ettore Di Giacinto <mudler@localai.io> * refactor(nodes): give the backend client factory the node id Mechanical: the method gains a nodeID parameter and the eight test fakes are updated. No behaviour change. Assisted-by: Claude:claude-sonnet-5-5 Signed-off-by: Ettore Di Giacinto <mudler@localai.io> * refactor(nodes): drop the optional node-aware factory The node id is now in the main method, so the optional interface and its helper had no behaviour of their own. Assisted-by: Claude:claude-sonnet-5-5 Signed-off-by: Ettore Di Giacinto <mudler@localai.io> * refactor(nodes): dial backend probes through the client factory Assisted-by: Claude:claude-sonnet-5-5 Signed-off-by: Ettore Di Giacinto <mudler@localai.io> * refactor(nodes): dial workers' file servers through a per-node dialer Assisted-by: Claude:claude-sonnet-5-5 Signed-off-by: Ettore Di Giacinto <mudler@localai.io> * refactor(http): proxy backend logs through the per-node worker dialer The admin backend-logs proxy (list, lines and the WebSocket stream) now reaches a worker through the same per-node dialer as the HTTP file stager, so every frontend-to-worker dial goes through one seam. The shared direct dialer keeps alive for 15s where the proxy used 30s. Harmless for requests bounded at 15s. Assisted-by: Claude:claude-sonnet-5-5 Signed-off-by: Ettore Di Giacinto <mudler@localai.io> * fix(http): keep the backend-logs proxy independent of the admin connection The proxy request had no context before the dialer change and is bounded only by its 15s timeout. Keep it that way. Assisted-by: Claude:claude-sonnet-5-5 Signed-off-by: Ettore Di Giacinto <mudler@localai.io> * refactor: move the worker control payloads to workerctl Mechanical move of the request and reply structs, the install progress event and the file payloads out of messaging. The verbs no longer belong to one carrier. No alias is left behind. Assisted-by: Claude:claude-sonnet-5-5 Signed-off-by: Ettore Di Giacinto <mudler@localai.io> * refactor(worker): serve the lifecycle verbs through a controlServer The worker registers one handler per verb and a NATS server maps each verb to its subject. Registration errors now name the verb. node.stop is served with SubscribeReply, which is identical on the wire because the handler never replies. Assisted-by: Claude:claude-sonnet-5-5 Signed-off-by: Ettore Di Giacinto <mudler@localai.io> * refactor(worker): report install progress through the control sink Install and upgrade now emit download progress through the sink the control server hands them. The debounce and the terminal flush stay in the handler path, built over that sink by the new nodes.NewDebouncedInstallProgressSink, which replaces NewDebouncedInstallProgressPublisher. The subject and payload on the wire are unchanged. The supervisor no longer holds the bus, and installFn and upgradeFn let specs drive both verbs without a gallery. The malformed-request log lines are restored for install, upgrade, backend.delete, model.unload, model.stop and model.delete, with the reply bytes unchanged. The signal adapter is renamed noReply, which also lets worker.go import os/signal without an alias again. Assisted-by: Claude:claude-sonnet-5-5 Signed-off-by: Ettore Di Giacinto <mudler@localai.io> * refactor(worker): serve the file-staging verbs through a controlServer An empty list-dir answer is now {} rather than {"files":null}, because the typed reply omits an empty Files slice. The frontend decodes both to a nil slice in nodes/file_stager_s3.go. Assisted-by: Claude:claude-sonnet-5-5 Signed-off-by: Ettore Di Giacinto <mudler@localai.io> * feat(messaging): add WorkQueue and the NATS producer This is the producer side of the competing-consumer seam. The work kinds map one to one to today's subjects and queue groups: task to jobs.new and mcp-ci to jobs.mcp-ci.new (both in group workers), agent-run to agent.execute (group agent-workers). Enqueue publishes the payload as Publish does today, with one JSON marshal. FakeBus now records queue groups and keeps reply handlers so later specs can pin and drive them. Assisted-by: Claude:claude-sonnet-5-5 Signed-off-by: Ettore Di Giacinto <mudler@localai.io> * feat(messaging): add the NATS WorkConsumer An in-flight limit of one runs the handler inline on the delivery goroutine, as the MCP CI consumer does today. Any other limit spawns per delivery, as the agent consumer does. Queue groups are unchanged. Assisted-by: Claude:claude-sonnet-5-5 Signed-off-by: Ettore Di Giacinto <mudler@localai.io> * refactor: publish queued work through WorkQueue The job dispatcher, the agent pool and the agent scheduler enqueue through messaging.WorkQueue; the NATS implementation publishes to the same subjects as before. DistributedServices builds the queue next to the NATS client and hands it to the dispatcher and the agent pool, whose distributed mode switch now reads a non-nil WorkQueue. The unused AgentPoolService.SetNATSClient is removed. Assisted-by: Claude:claude-sonnet-5-5 Signed-off-by: Ettore Di Giacinto <mudler@localai.io> * refactor: consume queued work through WorkConsumer The agent dispatcher and the MCP CI consumer register through messaging.WorkConsumer. The NATS implementation keeps the inline one-at-a-time model for MCP CI and the per-delivery model for agent runs. handleMCPCIJob reports on the events publisher the carrier hands it instead of a captured client. Assisted-by: Claude:claude-sonnet-5-5 Signed-off-by: Ettore Di Giacinto <mudler@localai.io> * refactor: delete the consumers nothing in production reached jobs.new has a producer and no production consumer, and the agent dispatcher's Dispatch was only called from tests. Publishing jobs.new is unchanged. Assisted-by: Claude:claude-sonnet-5-5 Signed-off-by: Ettore Di Giacinto <mudler@localai.io> * refactor(mcp): send MCP requests to agent workers through AgentControl Timeouts still honour only the deadline, not cancellation, exactly as today. The NATS no-responders error maps to ErrNoRoute and a timeout does not. Assisted-by: Claude:claude-sonnet-5-5 Signed-off-by: Ettore Di Giacinto <mudler@localai.io> * refactor(agent-worker): serve MCP requests and backend.stop through agentRPCServer The agent worker's MCP tool and discovery reply subscriptions and its backend stop listener move behind an unexported agentRPCServer interface, served on NATS by nodes.NATSAgentRPCServer. The handlers become typed mcp.ToolHandler and mcp.DiscoveryHandler values that answer every failure with a reply carrying Error. Queue group (agent-workers), inline execution on the delivery goroutine, the background handler context, the unmarshal error reply texts and the reply-less backend stop subscription are unchanged. The backend stop handler takes the decoded backend name, so it can still close that backend's MCP sessions. Assisted-by: Claude:claude-sonnet-5-5 Signed-off-by: Ettore Di Giacinto <mudler@localai.io> * refactor(messaging): remove helpers that only tests used BroadcastRoots, ControlRoots and SubjectRoot had no production caller. The roots spec now asserts every served root through ValidateSubject instead. MatchSubject moves back into the test support package, the only place that used it, with its table. NATSAgentRPCServer drops the subscription list it stored and never read, and NewNATSAgentRPCServer gets a doc comment. Assisted-by: Claude:claude-sonnet-5-5 Signed-off-by: Ettore Di Giacinto <mudler@localai.io> * test(mcp): round trip the agent RPC server over a real NATS server One spec sends a tool request and a discovery request through NATSAgentControl to NATSAgentRPCServer and checks that the handlers see the decoded requests and the replies come back. It also puts an undecodable body on the tool subject and checks the server answers with an unmarshal error instead of leaving the requester to time out. Assisted-by: Claude:claude-sonnet-5-5 Signed-off-by: Ettore Di Giacinto <mudler@localai.io> * docs: describe the distributed transport seams The developer note now lists the final seams: fan-out, queues, both halves of the control verbs and of agent RPC, and the dial. It records the open items a second carrier has to handle. Assisted-by: Claude:claude-sonnet-5-5 Signed-off-by: Ettore Di Giacinto <mudler@localai.io> * test: pin the in-flight limit each queue consumer asks for The work queue specs pin what Consume does for a given limit, but nothing pinned which limit each production consumer passes. Changing the agent worker's MCP CI limit from 1 to 0 would have let MCP CI jobs run concurrently on each worker with every test green. Move the MCP CI Consume call into startMCPCIConsumer with the same wiring and pin that it asks for (WorkMCPCI, 1). Pin that NATSDispatcher.Start asks for (WorkAgentRun, maxConcurrent) for several limits. Assisted-by: Claude:claude-sonnet-5-5 Signed-off-by: Ettore Di Giacinto <mudler@localai.io> * refactor: remove helpers the branch left without a caller SubjectJobCancelWildcard lost its last subscriber when the frontend stopped listening on jobs.*.cancel; the NATS permissions and conformance suite spell the subject out, so nothing reads the constant. decodeBackendStopRequest returned a stopAll flag that production dropped and only a test read. decodeBackendStop is now the single decoder with the same semantics: an empty body is stop-all, an empty Backend is stop-all, malformed JSON is an error. stopBackends still derives stop-all from Backend, so no reply changes. Assisted-by: Claude:claude-sonnet-5-5 Signed-off-by: Ettore Di Giacinto <mudler@localai.io> * fix(messaging): keep an explicitly empty agent queue a plain subscription Before the work queue seam the agent worker passed LOCALAI_AGENT_QUEUE straight to QueueSubscribe, so an explicitly empty value made a plain subscription and every agent worker ran every agent run. WithAgentRunRoute replaced an empty queue with agent-workers, which silently changed that. Keep the queue as given once the option is applied. An empty subject still falls back to agent.execute, since it never had a meaning of its own. The flag default stays agent-workers, so only an explicitly empty value reaches this. Assisted-by: Claude:claude-sonnet-5-5 Signed-off-by: Ettore Di Giacinto <mudler@localai.io> * docs: correct comments and record the PR B notes Fix the recordingFactory comment (it also records the parallel flag), document that a negative maxInFlight is unbounded and that Unsubscribe from a handler deadlocks, and say a permanently undecodable payload returns nil. Record controlHandler's undecodable return as a kept exception, and add the second carrier notes to the developer note: the reconciler has no ClientFactory option, the logs proxy honours HTTP_PROXY, verbs one carrier serves need an opt-out, terminal replies come from the result event, and agent runs publish through the NATS-bound EventBridge, which is not an additive change. Assisted-by: Claude:claude-sonnet-5-5 Signed-off-by: Ettore Di Giacinto <mudler@localai.io> --------- Signed-off-by: Ettore Di Giacinto <mudler@localai.io> Co-authored-by: Ettore Di Giacinto <mudler@localai.io>
568 lines
23 KiB
Go
568 lines
23 KiB
Go
package config
|
|
|
|
import (
|
|
"cmp"
|
|
"fmt"
|
|
"time"
|
|
|
|
"github.com/mudler/LocalAI/core/services/messaging"
|
|
"github.com/mudler/LocalAI/pkg/natsauth"
|
|
"github.com/mudler/xlog"
|
|
)
|
|
|
|
// DistributedConfig holds configuration for horizontal scaling mode.
|
|
// When Enabled is true, PostgreSQL and NATS are required.
|
|
type DistributedConfig struct {
|
|
Enabled bool // --distributed / LOCALAI_DISTRIBUTED
|
|
InstanceID string // --instance-id / LOCALAI_INSTANCE_ID (auto-generated UUID if empty)
|
|
NatsURL string // --nats-url / LOCALAI_NATS_URL
|
|
StorageURL string // --storage-url / LOCALAI_STORAGE_URL (S3 endpoint)
|
|
RegistrationToken string // --registration-token / LOCALAI_REGISTRATION_TOKEN (required token for node registration)
|
|
// RegistrationRequireAuth fails startup when distributed mode is enabled but
|
|
// RegistrationToken is empty. The default (false) keeps the historical
|
|
// fail-open behavior with a loud warning; production should set it so the
|
|
// node-register endpoints and the worker file-transfer server cannot run
|
|
// unauthenticated. Mirrors NatsRequireAuth for the NATS bus.
|
|
RegistrationRequireAuth bool // LOCALAI_REGISTRATION_REQUIRE_AUTH
|
|
// RequireAuth is the umbrella switch (LOCALAI_DISTRIBUTED_REQUIRE_AUTH) for
|
|
// distributed-mode auth: when true it implies BOTH NatsRequireAuth and
|
|
// RegistrationRequireAuth, so a single knob locks down the bus and the
|
|
// registration/file-transfer layer together. The granular flags remain
|
|
// available to enforce just one layer.
|
|
RequireAuth bool // LOCALAI_DISTRIBUTED_REQUIRE_AUTH
|
|
AutoApproveNodes bool // --auto-approve-nodes / LOCALAI_AUTO_APPROVE_NODES (skip admin approval for new workers)
|
|
// SharedModels asserts that every node (frontend and workers) mounts the
|
|
// SAME models directory at the SAME path (e.g. a shared volume, as in
|
|
// docker-compose.distributed.yaml). When true, the router skips staging
|
|
// model files to workers entirely: the frontend's absolute model paths are
|
|
// already valid on the worker, so re-uploading them into a per-model
|
|
// subdirectory only re-downloads what is already present (#10556). Default
|
|
// false preserves the historical per-node staging behavior.
|
|
SharedModels bool // --distributed-shared-models / LOCALAI_DISTRIBUTED_SHARED_MODELS
|
|
|
|
// NATS JWT auth (optional; see pkg/natsauth and docs/features/distributed-mode.md)
|
|
NatsAccountSeed string // LOCALAI_NATS_ACCOUNT_SEED — account signing seed to mint per-node worker JWTs
|
|
NatsServiceJWT string // LOCALAI_NATS_SERVICE_JWT — user JWT for frontends / agent workers
|
|
NatsServiceSeed string // LOCALAI_NATS_SERVICE_SEED — signing seed paired with service JWT
|
|
NatsWorkerJWTTTL time.Duration // LOCALAI_NATS_WORKER_JWT_TTL — minted worker JWT lifetime (default 24h)
|
|
NatsRequireAuth bool // LOCALAI_NATS_REQUIRE_AUTH — fail startup if NATS credentials are missing
|
|
NatsTLSCA string // LOCALAI_NATS_TLS_CA — PEM file for private CA (server verify)
|
|
NatsTLSCert string // LOCALAI_NATS_TLS_CERT — client cert for NATS mTLS
|
|
NatsTLSKey string // LOCALAI_NATS_TLS_KEY — client key paired with NatsTLSCert
|
|
|
|
// S3 configuration (used when StorageURL is set)
|
|
StorageBucket string // --storage-bucket / LOCALAI_STORAGE_BUCKET
|
|
StorageRegion string // --storage-region / LOCALAI_STORAGE_REGION
|
|
StorageAccessKey string // --storage-access-key / LOCALAI_STORAGE_ACCESS_KEY
|
|
StorageSecretKey string // --storage-secret-key / LOCALAI_STORAGE_SECRET_KEY
|
|
|
|
// Timeout configuration (all have sensible defaults — zero means use default)
|
|
MCPToolTimeout time.Duration // MCP tool execution timeout (default 360s)
|
|
MCPDiscoveryTimeout time.Duration // MCP discovery timeout (default 60s)
|
|
WorkerWaitTimeout time.Duration // Max wait for healthy worker at startup (default 5m)
|
|
DrainTimeout time.Duration // Time to wait for in-flight requests during drain (default 30s)
|
|
HealthCheckInterval time.Duration // Health monitor check interval (default 15s)
|
|
StaleNodeThreshold time.Duration // Time before a node is considered stale (default 5m)
|
|
NodeHeartbeatCheckpoint time.Duration // Minimum gap between durable heartbeat writes (default 60s, 0 = every beat)
|
|
// DisablePerModelHealthCheck turns off the health monitor's per-model
|
|
// gRPC probe. When enabled (the default), the monitor pings each model's
|
|
// gRPC address and removes stale node_models rows whose backend has
|
|
// crashed even though the worker's node-level heartbeat is still arriving.
|
|
// Without per-model probing, /embeddings and /completions can be dispatched
|
|
// to a backend that silently returns garbage (see also the cascading
|
|
// model-row cleanup on MarkUnhealthy / MarkDraining).
|
|
DisablePerModelHealthCheck bool
|
|
|
|
MCPCIJobTimeout time.Duration // MCP CI job execution timeout (default 10m)
|
|
|
|
BackendInstallTimeout time.Duration // NATS round-trip timeout for backend.install (default 15m)
|
|
BackendUpgradeTimeout time.Duration // NATS round-trip timeout for backend.upgrade (default 15m)
|
|
// ModelLoadTimeout is the gRPC deadline for the remote LoadModel call the
|
|
// router issues once a worker has the backend installed and the model files
|
|
// staged. It therefore covers only the backend's own checkpoint load and
|
|
// pipeline init, which for a multi-tens-of-GB diffusion/video checkpoint on
|
|
// unified memory can far exceed the 5m default.
|
|
ModelLoadTimeout time.Duration // gRPC deadline for remote LoadModel (default 5m)
|
|
// ModelLoadWait bounds how long an inference request waits for a model that
|
|
// is being cold-loaded before it is answered with 503 plus live progress. A
|
|
// held HTTP request cannot survive real infrastructure — an ingress or LB
|
|
// idle timeout kills a 20-minute request regardless of what LocalAI does —
|
|
// so the wait is bounded by default.
|
|
//
|
|
// Zero means unset (DefaultModelLoadWait applies); ModelLoadWaitUnbounded
|
|
// records the operator asking for unbounded waiting with
|
|
// LOCALAI_MODEL_LOAD_WAIT=0.
|
|
ModelLoadWait time.Duration
|
|
|
|
MaxUploadSize int64 // Maximum upload body size in bytes (default 50 GB)
|
|
|
|
AgentWorkerConcurrency int `yaml:"agent_worker_concurrency" json:"agent_worker_concurrency" env:"LOCALAI_AGENT_WORKER_CONCURRENCY"`
|
|
|
|
// DiskHeadroomDisabled turns off the scheduler's free-disk admission check,
|
|
// restoring the pre-#11054 behaviour where node selection ignores whether a
|
|
// node can actually store the model. The check is ON by default because it
|
|
// prevents a measured failure (a node with 0 bytes free accepted a 70GB
|
|
// model and failed 16 minutes into staging); this is the escape hatch for
|
|
// setups where our size estimate is wrong (deduplicating filesystems, a
|
|
// worker that fetches its own weights), not the norm.
|
|
//
|
|
// Disabling does NOT silence the check: it still runs and warns when it
|
|
// would have rejected every node, so the operator keeps the diagnosis
|
|
// without being blocked. See SmartRouter.scheduleNewModel.
|
|
//
|
|
// Stored as the negation of the CLI/runtime flag so the zero value is
|
|
// "enabled" (mirrors PrefixCacheDisabled).
|
|
DiskHeadroomDisabled bool
|
|
|
|
// PrefixCacheDisabled turns off prefix-cache-aware routing, falling back to
|
|
// round-robin (the floor). Prefix-cache routing is ON by default in
|
|
// distributed mode; this flag exists so operators can opt out. The CLI
|
|
// surfaces a default-true --distributed-prefix-cache enable flag and sets
|
|
// this when the operator passes --distributed-prefix-cache=false.
|
|
PrefixCacheDisabled bool
|
|
// PrefixCacheTTL is the idle-timeout for prefix-cache index entries and
|
|
// drives the background eviction cadence (eviction runs every TTL/2). Zero
|
|
// means use the prefixcache package default (5m).
|
|
PrefixCacheTTL time.Duration
|
|
// ModelSchedulingJSON is an inline JSON list of per-model scheduling configs
|
|
// applied authoritatively at startup (LOCALAI_MODEL_SCHEDULING).
|
|
ModelSchedulingJSON string
|
|
// ModelSchedulingConfigPath is a path to a YAML file with the same list
|
|
// (LOCALAI_MODEL_SCHEDULING_CONFIG).
|
|
ModelSchedulingConfigPath string
|
|
}
|
|
|
|
// Validate checks that the distributed configuration is internally consistent.
|
|
// It returns nil if distributed mode is disabled.
|
|
func (c DistributedConfig) Validate() error {
|
|
if !c.Enabled {
|
|
return nil
|
|
}
|
|
if c.NatsURL == "" {
|
|
return fmt.Errorf("distributed mode requires --nats-url / LOCALAI_NATS_URL")
|
|
}
|
|
// S3 credentials must be paired
|
|
if (c.StorageAccessKey != "" && c.StorageSecretKey == "") ||
|
|
(c.StorageAccessKey == "" && c.StorageSecretKey != "") {
|
|
return fmt.Errorf("storage-access-key and storage-secret-key must both be set or both empty")
|
|
}
|
|
// The registration token guards both the node HTTP register/heartbeat
|
|
// endpoints and the worker file-transfer server (which fails open on an
|
|
// empty token). Enforce it when registration auth is required (the granular
|
|
// flag or the umbrella); otherwise warn.
|
|
if c.RegistrationToken == "" {
|
|
if c.RegistrationAuthRequired() {
|
|
return fmt.Errorf("registration auth is required (LOCALAI_REGISTRATION_REQUIRE_AUTH or LOCALAI_DISTRIBUTED_REQUIRE_AUTH) but LOCALAI_REGISTRATION_TOKEN is empty")
|
|
}
|
|
xlog.Warn("distributed mode running without registration token — node endpoints and the worker file-transfer server are unprotected; set LOCALAI_REGISTRATION_TOKEN, or LOCALAI_DISTRIBUTED_REQUIRE_AUTH=true to fail closed")
|
|
}
|
|
if err := c.NatsAuthConfig().Validate(); err != nil {
|
|
return err
|
|
}
|
|
if err := c.NatsTLSFiles().Validate(); err != nil {
|
|
return err
|
|
}
|
|
c.NatsAuthConfig().WarnIfInsecure(true)
|
|
// Check for negative durations
|
|
for name, d := range map[string]time.Duration{
|
|
FlagMCPToolTimeout: c.MCPToolTimeout,
|
|
FlagMCPDiscoveryTimeout: c.MCPDiscoveryTimeout,
|
|
FlagWorkerWaitTimeout: c.WorkerWaitTimeout,
|
|
FlagDrainTimeout: c.DrainTimeout,
|
|
FlagHealthCheckInterval: c.HealthCheckInterval,
|
|
FlagStaleNodeThreshold: c.StaleNodeThreshold,
|
|
FlagNodeHeartbeatCheckpoint: c.NodeHeartbeatCheckpoint,
|
|
FlagMCPCIJobTimeout: c.MCPCIJobTimeout,
|
|
FlagBackendInstallTimeout: c.BackendInstallTimeout,
|
|
FlagBackendUpgradeTimeout: c.BackendUpgradeTimeout,
|
|
FlagModelLoadTimeout: c.ModelLoadTimeout,
|
|
} {
|
|
if d < 0 {
|
|
return fmt.Errorf("%s must not be negative", name)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// Distributed config options
|
|
|
|
var EnableDistributed = func(o *ApplicationConfig) {
|
|
o.Distributed.Enabled = true
|
|
}
|
|
|
|
func WithDistributedInstanceID(id string) AppOption {
|
|
return func(o *ApplicationConfig) {
|
|
o.Distributed.InstanceID = id
|
|
}
|
|
}
|
|
|
|
func WithNatsURL(url string) AppOption {
|
|
return func(o *ApplicationConfig) {
|
|
o.Distributed.NatsURL = url
|
|
}
|
|
}
|
|
|
|
func WithRegistrationToken(token string) AppOption {
|
|
return func(o *ApplicationConfig) {
|
|
o.Distributed.RegistrationToken = token
|
|
}
|
|
}
|
|
|
|
func WithNatsAccountSeed(seed string) AppOption {
|
|
return func(o *ApplicationConfig) {
|
|
o.Distributed.NatsAccountSeed = seed
|
|
}
|
|
}
|
|
|
|
func WithNatsServiceJWT(jwt string) AppOption {
|
|
return func(o *ApplicationConfig) {
|
|
o.Distributed.NatsServiceJWT = jwt
|
|
}
|
|
}
|
|
|
|
func WithNatsServiceSeed(seed string) AppOption {
|
|
return func(o *ApplicationConfig) {
|
|
o.Distributed.NatsServiceSeed = seed
|
|
}
|
|
}
|
|
|
|
func WithNatsWorkerJWTTTL(d time.Duration) AppOption {
|
|
return func(o *ApplicationConfig) {
|
|
o.Distributed.NatsWorkerJWTTTL = d
|
|
}
|
|
}
|
|
|
|
var EnableNatsRequireAuth = func(o *ApplicationConfig) {
|
|
o.Distributed.NatsRequireAuth = true
|
|
}
|
|
|
|
// EnableRegistrationRequireAuth makes an empty registration token a hard error
|
|
// in distributed mode (see DistributedConfig.RegistrationRequireAuth).
|
|
var EnableRegistrationRequireAuth = func(o *ApplicationConfig) {
|
|
o.Distributed.RegistrationRequireAuth = true
|
|
}
|
|
|
|
// EnableDistributedRequireAuth is the umbrella switch implying both
|
|
// NatsRequireAuth and RegistrationRequireAuth (see DistributedConfig.RequireAuth).
|
|
var EnableDistributedRequireAuth = func(o *ApplicationConfig) {
|
|
o.Distributed.RequireAuth = true
|
|
}
|
|
|
|
// RegistrationAuthRequired reports whether an empty registration token must be
|
|
// treated as a fatal misconfiguration — the granular flag or the umbrella.
|
|
func (c DistributedConfig) RegistrationAuthRequired() bool {
|
|
return c.RegistrationRequireAuth || c.RequireAuth
|
|
}
|
|
|
|
// NatsAuthRequired reports whether NATS JWT credentials must be present — the
|
|
// granular flag or the umbrella.
|
|
func (c DistributedConfig) NatsAuthRequired() bool {
|
|
return c.NatsRequireAuth || c.RequireAuth
|
|
}
|
|
|
|
func WithNatsTLSCA(path string) AppOption {
|
|
return func(o *ApplicationConfig) {
|
|
o.Distributed.NatsTLSCA = path
|
|
}
|
|
}
|
|
|
|
func WithNatsTLSCert(path string) AppOption {
|
|
return func(o *ApplicationConfig) {
|
|
o.Distributed.NatsTLSCert = path
|
|
}
|
|
}
|
|
|
|
func WithNatsTLSKey(path string) AppOption {
|
|
return func(o *ApplicationConfig) {
|
|
o.Distributed.NatsTLSKey = path
|
|
}
|
|
}
|
|
|
|
func WithStorageURL(url string) AppOption {
|
|
return func(o *ApplicationConfig) {
|
|
o.Distributed.StorageURL = url
|
|
}
|
|
}
|
|
|
|
func WithStorageBucket(bucket string) AppOption {
|
|
return func(o *ApplicationConfig) {
|
|
o.Distributed.StorageBucket = bucket
|
|
}
|
|
}
|
|
|
|
func WithStorageRegion(region string) AppOption {
|
|
return func(o *ApplicationConfig) {
|
|
o.Distributed.StorageRegion = region
|
|
}
|
|
}
|
|
|
|
func WithStorageAccessKey(key string) AppOption {
|
|
return func(o *ApplicationConfig) {
|
|
o.Distributed.StorageAccessKey = key
|
|
}
|
|
}
|
|
|
|
func WithStorageSecretKey(key string) AppOption {
|
|
return func(o *ApplicationConfig) {
|
|
o.Distributed.StorageSecretKey = key
|
|
}
|
|
}
|
|
|
|
func WithBackendInstallTimeout(d time.Duration) AppOption {
|
|
return func(o *ApplicationConfig) {
|
|
o.Distributed.BackendInstallTimeout = d
|
|
}
|
|
}
|
|
|
|
func WithBackendUpgradeTimeout(d time.Duration) AppOption {
|
|
return func(o *ApplicationConfig) {
|
|
o.Distributed.BackendUpgradeTimeout = d
|
|
}
|
|
}
|
|
|
|
func WithModelLoadTimeout(d time.Duration) AppOption {
|
|
return func(o *ApplicationConfig) {
|
|
o.Distributed.ModelLoadTimeout = d
|
|
}
|
|
}
|
|
|
|
// WithModelLoadWait sets how long a request waits for a cold-loading model. A
|
|
// zero d records the operator asking for unbounded waiting: "set the knob to
|
|
// zero" cannot sensibly mean "use the default".
|
|
func WithModelLoadWait(d time.Duration) AppOption {
|
|
return func(o *ApplicationConfig) {
|
|
if d == 0 {
|
|
d = ModelLoadWaitUnbounded
|
|
}
|
|
o.Distributed.ModelLoadWait = d
|
|
}
|
|
}
|
|
|
|
// WithStaleNodeThreshold sets how long a node may go without a durable
|
|
// heartbeat before the health monitor marks it offline. It has to be raised
|
|
// alongside WithNodeHeartbeatCheckpoint: a checkpoint interval wider than this
|
|
// threshold makes every healthy node look dead the moment its beats start
|
|
// being suppressed.
|
|
func WithStaleNodeThreshold(d time.Duration) AppOption {
|
|
return func(o *ApplicationConfig) {
|
|
o.Distributed.StaleNodeThreshold = d
|
|
}
|
|
}
|
|
|
|
// WithNodeHeartbeatCheckpoint bounds durable heartbeat writes. A zero d is
|
|
// deliberately not special-cased into "unbounded": NodeHeartbeatCheckpointOrDefault
|
|
// reads zero as unset, and an operator who wants a write per beat sets a value
|
|
// below the worker's heartbeat interval instead.
|
|
func WithNodeHeartbeatCheckpoint(d time.Duration) AppOption {
|
|
return func(o *ApplicationConfig) {
|
|
o.Distributed.NodeHeartbeatCheckpoint = d
|
|
}
|
|
}
|
|
|
|
var EnableAutoApproveNodes = func(o *ApplicationConfig) {
|
|
o.Distributed.AutoApproveNodes = true
|
|
}
|
|
|
|
// EnableDistributedSharedModels marks the cluster as sharing one models
|
|
// directory across all nodes, so the router skips staging model files to
|
|
// workers (see DistributedConfig.SharedModels).
|
|
var EnableDistributedSharedModels = func(o *ApplicationConfig) {
|
|
o.Distributed.SharedModels = true
|
|
}
|
|
|
|
// DisableDiskHeadroomCheck turns off the scheduler's free-disk admission
|
|
// check (see DistributedConfig.DiskHeadroomDisabled). The check is enabled by
|
|
// default in distributed mode.
|
|
var DisableDiskHeadroomCheck = func(o *ApplicationConfig) {
|
|
o.Distributed.DiskHeadroomDisabled = true
|
|
}
|
|
|
|
// DisablePrefixCache turns off prefix-cache-aware routing (falls back to
|
|
// round-robin). Prefix-cache routing is enabled by default in distributed mode.
|
|
var DisablePrefixCache = func(o *ApplicationConfig) {
|
|
o.Distributed.PrefixCacheDisabled = true
|
|
}
|
|
|
|
// WithPrefixCacheTTL sets the prefix-cache index idle-timeout (and the
|
|
// background eviction cadence, which runs every TTL/2).
|
|
func WithPrefixCacheTTL(d time.Duration) AppOption {
|
|
return func(o *ApplicationConfig) {
|
|
o.Distributed.PrefixCacheTTL = d
|
|
}
|
|
}
|
|
|
|
// WithModelSchedulingJSON sets the inline-JSON declarative scheduling config.
|
|
func WithModelSchedulingJSON(s string) AppOption {
|
|
return func(o *ApplicationConfig) {
|
|
o.Distributed.ModelSchedulingJSON = s
|
|
}
|
|
}
|
|
|
|
// WithModelSchedulingConfigPath sets the path to a YAML declarative scheduling
|
|
// config file.
|
|
func WithModelSchedulingConfigPath(path string) AppOption {
|
|
return func(o *ApplicationConfig) {
|
|
o.Distributed.ModelSchedulingConfigPath = path
|
|
}
|
|
}
|
|
|
|
// Flag names for distributed timeout / interval configuration. These are
|
|
// the kebab-case identifiers kong derives from the matching RunCMD struct
|
|
// fields; they appear in Validate error messages and any other operator-
|
|
// facing surface that needs to reference a specific knob by name. Keeping
|
|
// them as constants prevents the string from drifting from the actual
|
|
// flag a future rename would produce.
|
|
const (
|
|
FlagMCPToolTimeout = "mcp-tool-timeout"
|
|
FlagMCPDiscoveryTimeout = "mcp-discovery-timeout"
|
|
FlagWorkerWaitTimeout = "worker-wait-timeout"
|
|
FlagDrainTimeout = "drain-timeout"
|
|
FlagHealthCheckInterval = "health-check-interval"
|
|
FlagStaleNodeThreshold = "stale-node-threshold"
|
|
FlagNodeHeartbeatCheckpoint = "node-heartbeat-checkpoint"
|
|
FlagMCPCIJobTimeout = "mcp-ci-job-timeout"
|
|
FlagBackendInstallTimeout = "backend-install-timeout"
|
|
FlagBackendUpgradeTimeout = "backend-upgrade-timeout"
|
|
FlagModelLoadTimeout = "model-load-timeout"
|
|
FlagModelLoadWait = "model-load-wait"
|
|
// FlagDiskHeadroomCheck names the disk-headroom toggle. It is quoted in
|
|
// the warning the check emits while disabled, so the operator reading a
|
|
// log line knows exactly which knob produced it.
|
|
FlagDiskHeadroomCheck = "distributed-disk-headroom-check"
|
|
)
|
|
|
|
// Defaults for distributed timeouts.
|
|
const (
|
|
DefaultMCPToolTimeout = 360 * time.Second
|
|
DefaultMCPDiscoveryTimeout = 60 * time.Second
|
|
DefaultWorkerWaitTimeout = 5 * time.Minute
|
|
DefaultDrainTimeout = 30 * time.Second
|
|
DefaultHealthCheckInterval = 15 * time.Second
|
|
// A beat that only refreshes the timestamp is now dropped until the
|
|
// checkpoint interval elapses, so the persisted column is up to one
|
|
// interval stale by design. The threshold covers that plus jitter.
|
|
// A genuinely dead node is still caught sooner by the per-model gRPC
|
|
// health check and by request-time failure, neither of which reads this.
|
|
DefaultStaleNodeThreshold = 5 * time.Minute
|
|
DefaultNodeHeartbeatCheckpoint = 60 * time.Second
|
|
DefaultMCPCIJobTimeout = 10 * time.Minute
|
|
DefaultBackendInstallTimeout = 15 * time.Minute
|
|
DefaultBackendUpgradeTimeout = 15 * time.Minute
|
|
DefaultModelLoadTimeout = 5 * time.Minute
|
|
// DefaultModelLoadWait is how long a request waits for a cold-loading model
|
|
// before it is answered with 503 and live progress. Chosen to sit under the
|
|
// idle timeout of typical ingress/LB defaults, so the answer comes from
|
|
// LocalAI (with progress the client can act on) rather than from a proxy
|
|
// dropping the connection.
|
|
DefaultModelLoadWait = 60 * time.Second
|
|
)
|
|
|
|
// ModelLoadWaitUnbounded records LOCALAI_MODEL_LOAD_WAIT=0 — "wait as long as
|
|
// it takes" — which a plain zero cannot express, since zero also means "unset,
|
|
// use the default". Only deployments with no proxy in front should use it.
|
|
const ModelLoadWaitUnbounded = -1 * time.Second
|
|
|
|
// DefaultMaxUploadSize is the default maximum upload body size (50 GB).
|
|
const DefaultMaxUploadSize int64 = 50 << 30
|
|
|
|
// NatsTLSFiles returns NATS TLS/mTLS PEM paths for the messaging client.
|
|
func (c DistributedConfig) NatsTLSFiles() messaging.TLSFiles {
|
|
return messaging.TLSFiles{
|
|
CA: c.NatsTLSCA,
|
|
Cert: c.NatsTLSCert,
|
|
Key: c.NatsTLSKey,
|
|
}
|
|
}
|
|
|
|
// NatsMessagingOptions builds messaging client options (JWT + TLS) for distributed components.
|
|
// Pass explicit userJWT/userSeed when set (e.g. worker overrides); empty uses service JWT from config.
|
|
func (c DistributedConfig) NatsMessagingOptions(userJWT, userSeed string) []messaging.Option {
|
|
var opts []messaging.Option
|
|
jwt, seed := userJWT, userSeed
|
|
if jwt == "" && seed == "" {
|
|
auth := c.NatsAuthConfig()
|
|
jwt, seed = auth.ServiceUserJWT, auth.ServiceUserSeed
|
|
}
|
|
if jwt != "" && seed != "" {
|
|
opts = append(opts, messaging.WithUserJWT(jwt, seed))
|
|
}
|
|
if tls := c.NatsTLSFiles(); tls.Enabled() {
|
|
opts = append(opts, messaging.WithTLS(tls))
|
|
}
|
|
return opts
|
|
}
|
|
|
|
// NatsAuthConfig builds pkg/natsauth settings from distributed configuration.
|
|
func (c DistributedConfig) NatsAuthConfig() natsauth.Config {
|
|
return natsauth.Config{
|
|
AccountSeed: c.NatsAccountSeed,
|
|
ServiceUserJWT: c.NatsServiceJWT,
|
|
ServiceUserSeed: c.NatsServiceSeed,
|
|
WorkerJWTTTL: c.NatsWorkerJWTTTL,
|
|
RequireAuth: c.NatsAuthRequired(),
|
|
}
|
|
}
|
|
|
|
// BackendInstallTimeoutOrDefault returns the configured timeout or the default.
|
|
func (c DistributedConfig) BackendInstallTimeoutOrDefault() time.Duration {
|
|
return cmp.Or(c.BackendInstallTimeout, DefaultBackendInstallTimeout)
|
|
}
|
|
|
|
// BackendUpgradeTimeoutOrDefault returns the configured timeout or the default.
|
|
func (c DistributedConfig) BackendUpgradeTimeoutOrDefault() time.Duration {
|
|
return cmp.Or(c.BackendUpgradeTimeout, DefaultBackendUpgradeTimeout)
|
|
}
|
|
|
|
// ModelLoadTimeoutOrDefault returns the configured timeout or the default.
|
|
func (c DistributedConfig) ModelLoadTimeoutOrDefault() time.Duration {
|
|
return cmp.Or(c.ModelLoadTimeout, DefaultModelLoadTimeout)
|
|
}
|
|
|
|
// MCPToolTimeoutOrDefault returns the configured timeout or the default.
|
|
func (c DistributedConfig) MCPToolTimeoutOrDefault() time.Duration {
|
|
return cmp.Or(c.MCPToolTimeout, DefaultMCPToolTimeout)
|
|
}
|
|
|
|
// MCPDiscoveryTimeoutOrDefault returns the configured timeout or the default.
|
|
func (c DistributedConfig) MCPDiscoveryTimeoutOrDefault() time.Duration {
|
|
return cmp.Or(c.MCPDiscoveryTimeout, DefaultMCPDiscoveryTimeout)
|
|
}
|
|
|
|
// WorkerWaitTimeoutOrDefault returns the configured timeout or the default.
|
|
func (c DistributedConfig) WorkerWaitTimeoutOrDefault() time.Duration {
|
|
return cmp.Or(c.WorkerWaitTimeout, DefaultWorkerWaitTimeout)
|
|
}
|
|
|
|
// DrainTimeoutOrDefault returns the configured timeout or the default.
|
|
func (c DistributedConfig) DrainTimeoutOrDefault() time.Duration {
|
|
return cmp.Or(c.DrainTimeout, DefaultDrainTimeout)
|
|
}
|
|
|
|
// HealthCheckIntervalOrDefault returns the configured interval or the default.
|
|
func (c DistributedConfig) HealthCheckIntervalOrDefault() time.Duration {
|
|
return cmp.Or(c.HealthCheckInterval, DefaultHealthCheckInterval)
|
|
}
|
|
|
|
// StaleNodeThresholdOrDefault returns the configured threshold or the default.
|
|
func (c DistributedConfig) StaleNodeThresholdOrDefault() time.Duration {
|
|
return cmp.Or(c.StaleNodeThreshold, DefaultStaleNodeThreshold)
|
|
}
|
|
|
|
// NodeHeartbeatCheckpointOrDefault returns the configured interval or the
|
|
// default. A configured zero is indistinguishable from unset here, which is
|
|
// intentional: cmp.Or falls back to the default, and an operator who wants a
|
|
// write per beat sets a value below the heartbeat interval instead.
|
|
func (c DistributedConfig) NodeHeartbeatCheckpointOrDefault() time.Duration {
|
|
return cmp.Or(c.NodeHeartbeatCheckpoint, DefaultNodeHeartbeatCheckpoint)
|
|
}
|
|
|
|
// MCPCIJobTimeoutOrDefault returns the configured MCP CI job timeout or the default.
|
|
func (c DistributedConfig) MCPCIJobTimeoutOrDefault() time.Duration {
|
|
return cmp.Or(c.MCPCIJobTimeout, DefaultMCPCIJobTimeout)
|
|
}
|
|
|
|
// MaxUploadSizeOrDefault returns the configured max upload size or the default.
|
|
func (c DistributedConfig) MaxUploadSizeOrDefault() int64 {
|
|
return cmp.Or(c.MaxUploadSize, DefaultMaxUploadSize)
|
|
}
|