Documentation
¶
Overview ¶
Package executorstore backs the remote-executor enrollment flow with SQLite.
It is a type-converting adapter and nothing else: pkg/statedb owns the rows and the SQL, pkg/executor/remote owns the protocol and the crypto, and this package joins them. Keeping it separate keeps two dependencies apart that neither side wants:
- pkg/executor/remote must not depend on SQLite. The agent binary running on an edge device imports it, and linking a database engine into a process whose whole job is to fork workloads is dead weight on exactly the hardware that can least afford it.
- pkg/statedb must not depend on pkg/executor/remote, which pulls in a WebSocket stack a storage layer has no business knowing about.
The duplication of record shapes across the boundary is the price of that separation, and it is a small and stable price: these structs change only when the enrollment schema does.
Index ¶
- Constants
- func FailoverProjectPath(sess executor.Session) string
- func NewClaimToken() (string, error)
- func NewSessionID() (string, error)
- type FleetMember
- type HandleLoss
- type Handles
- func (h *Handles) ClaimHandleMeta(handleID, key, expect, value string) (bool, error)
- func (h *Handles) DeleteHandle(handleID string) error
- func (h *Handles) HandleMeta(handleID, key string) (string, bool, error)
- func (h *Handles) ListHandles(executorID string) ([]executor.HandleRecord, error)
- func (h *Handles) PutHandle(rec executor.HandleRecord) error
- type Records
- type Scheduler
- func (s *Scheduler) ClaimRequeue(sessionID, claimToken string, maxAttempts int, at time.Time) (executor.Session, bool, error)
- func (s *Scheduler) CloseSession(sessionID, state string, at time.Time) error
- func (s *Scheduler) CountRunning(executorID string) (int, error)
- func (s *Scheduler) ExecutorFailover(ev executor.FailoverEvent)
- func (s *Scheduler) ExecutorTransition(t executor.Transition)
- func (s *Scheduler) FailedOverProjects() ([]string, error)
- func (s *Scheduler) ForgetHealth(executorID string) error
- func (s *Scheduler) ListHealth() ([]executor.Health, error)
- func (s *Scheduler) LoadHealth(executorID string) (executor.Health, error)
- func (s *Scheduler) LossOfHandle(executorID, handleID string) (HandleLoss, error)
- func (s *Scheduler) LostNodes(sessionID string) ([]pm.NodeLoss, error)
- func (s *Scheduler) OpenRequeuedSession(sess executor.Session, requeuedFrom string) error
- func (s *Scheduler) OpenSession(sess executor.Session) error
- func (s *Scheduler) RunningSessions(executorID string) ([]executor.Session, error)
- func (s *Scheduler) RunningTasksOf(sessionID string) ([]int, error)
- func (s *Scheduler) SaveHealth(h executor.Health) error
- func (s *Scheduler) SessionState(sessionID string) (state string, replaced bool, err error)
- func (s *Scheduler) SetRunningTasks(sessionID string, tasks []int) (bool, error)
- func (s *Scheduler) WithActor(actor string) *Scheduler
- type Store
- func (s *Store) GetAgent(agentID string) (remote.AgentRecord, error)
- func (s *Store) GetEnrollment(id string) (remote.EnrollmentRecord, error)
- func (s *Store) ListAgents() ([]remote.AgentRecord, error)
- func (s *Store) ListEnrollments() ([]remote.EnrollmentRecord, error)
- func (s *Store) PutAgent(rec remote.AgentRecord) error
- func (s *Store) PutEnrollment(rec remote.EnrollmentRecord) error
- func (s *Store) RedeemEnrollment(id, agentID string, at time.Time) error
- func (s *Store) RevokeAgent(agentID string, at time.Time) error
- func (s *Store) RevokeEnrollment(id string, at time.Time) error
- func (s *Store) TouchAgent(agentID string, at time.Time) error
Constants ¶
const ( SourceAgentCredential = "agent credential" SourceConfiguration = "configuration" )
Record sources, for the messages that explain a match or a take-back.
const ( AuditEntityExecutor = "executor" AuditEventStateChange = auditaction.ActionExecutorStateChange AuditEventFailover = auditaction.ActionExecutorFailover AuditEventExhausted = auditaction.ActionExecutorFailoverExhausted AuditActorSupervisor = "supervisor" )
Audit constants for fleet events. EntityType is "executor" so an operator can filter the audit log down to infrastructure changes with one predicate.
Variables ¶
This section is empty.
Functions ¶
func FailoverProjectPath ¶ added in v0.0.5
FailoverProjectPath is the hub's own path to the project a session belongs to. A session dispatched to an executor that does not share the hub's filesystem records the executor's work directory as its ProjectPath — a path relative to the device's work root, meaningless on the hub — while the project label every hub dispatch carries names the hub's path (uiSpec). The label wins; the recorded path is the fallback for sessions dispatched without one.
func NewClaimToken ¶
NewClaimToken mints a session claim token.
The token arbitrates a race; it is not a secret, and anything that can write to the database can rewrite the row directly. crypto/rand is used anyway because it costs nothing here and removes the need to reason about whether a predictable token could ever let a stale supervisor guess its way back into ownership of a session it no longer holds.
func NewSessionID ¶
NewSessionID mints a session identifier. It is random rather than sequential so session IDs carry no information about fleet size or dispatch rate.
Types ¶
type FleetMember ¶ added in v0.0.7
type FleetMember struct {
ID string
Name string
Kind string
// Device is set for a virtual executor: the device whose labels it
// carries.
Device string
Record secretbroker.RecordedExecutor
}
FleetMember is one executor the hub knows, with its record, for asking which executors a label selector reaches (Task 20432).
func Fleet ¶ added in v0.0.7
func Fleet(db *statedb.DB, records secretbroker.ExecutorRecords, extra []FleetMember) ([]FleetMember, error)
Fleet lists every executor with a registry row — the devices, their virtual executors, the configured drivers — and every one in extra that has none, each with records' record of it, sorted by ID.
func Reaches ¶ added in v0.0.7
func Reaches(members []FleetMember, sub secretbroker.Subject) []FleetMember
Reaches returns the members a grant to sub would be delivered to now: for a label selector, those whose recorded labels carry every pair it names. A project subject's reach is not a set of executors, nor "any"'s; both give nil.
type HandleLoss ¶ added in v0.0.5
type HandleLoss struct {
// SessionID is the session the workload ran as, "" when none is recorded
// (session tracking is best-effort: a dispatch whose row could not be
// written runs untracked, and is never failed over).
SessionID string
// State is the session's recorded state.
State string
// Attempt is the session's dispatch count, 1 for a first dispatch.
Attempt int
// ClaimedAt is when the claim took the session, for a claimed one: the
// time its state last changed.
ClaimedAt time.Time
// Successor is the session a failover opened to replace this one, and
// SuccessorExecutor and SuccessorHandle where it runs; all "" when none.
Successor string
SuccessorExecutor string
SuccessorHandle string
}
HandleLoss is what the session store says about one dispatched workload, looked up by its executor and handle (Task 20396): whether the failover claim took it off its executor, and what replaced it.
It exists because the hub follows a run by its executor and handle, while the supervisor that decides the run's fate claims it by session — and is often in another process. The store is where the two meet.
func (HandleLoss) Claimed ¶ added in v0.0.5
func (l HandleLoss) Claimed() bool
Claimed reports whether the failover claim took the session off its executor: requeued (re-dispatched, or with nowhere to go) or failover_exhausted. A claimed session's workload is no longer the run of its project, whatever its driver still says about it.
func (HandleLoss) Exhausted ¶ added in v0.0.5
func (l HandleLoss) Exhausted() bool
Exhausted reports whether the claim closed it past the failover cap.
func (HandleLoss) Replaced ¶ added in v0.0.5
func (l HandleLoss) Replaced() bool
Replaced reports whether a failover started a replacement for it.
type Handles ¶
type Handles struct {
// contains filtered or unexported fields
}
Handles implements executor.HandleStore over a *statedb.DB.
It is a separate type from Scheduler rather than more methods on it because the two have different owners: a Scheduler is the control plane's view and is constructed by the supervisor, while a Handles is handed to a *driver*, which is the one thing in this system that may also run outside the hub. Keeping them apart means a driver holds a store that can do exactly three things and cannot, say, close a session it does not own.
func NewHandles ¶
NewHandles wraps a database handle.
func (*Handles) ClaimHandleMeta ¶ added in v0.0.5
ClaimHandleMeta sets one meta key of a handle's row from expect to value, conditionally (Task 20419). See statedb.ClaimExecutorHandleMeta.
func (*Handles) DeleteHandle ¶
DeleteHandle implements executor.HandleStore.
func (*Handles) HandleMeta ¶ added in v0.0.5
HandleMeta returns one meta key of a handle's row, and whether the row exists.
func (*Handles) ListHandles ¶
func (h *Handles) ListHandles(executorID string) ([]executor.HandleRecord, error)
ListHandles implements executor.HandleStore.
type Records ¶ added in v0.0.7
type Records struct {
// DB holds the agent credentials and the virtual executors.
DB *statedb.DB
// Configured returns the labels the hub's configuration gives each
// configured executor it enables, by ID (config.ExecutorsConfig.
// ConfiguredExecutorLabels). Asked on every lookup, so an edit to the
// configuration applies to the next lease. Nil for a process that
// configures no executor.
Configured func() (map[string]map[string]string, error)
}
Records implements secretbroker.ExecutorRecords over the control plane's database and the hub's configuration.
func (Records) RecordedExecutor ¶ added in v0.0.7
func (r Records) RecordedExecutor(executorID string) (secretbroker.RecordedExecutor, error)
RecordedExecutor implements secretbroker.ExecutorRecords.
A table or a configuration that cannot be read is an error. Answering "no labels" instead would dispatch the run without the label grants its executor holds — the silent non-delivery Task 20432 exists to end — and say nothing about why.
type Scheduler ¶
type Scheduler struct {
// contains filtered or unexported fields
}
Scheduler implements executor.HealthStore, executor.SessionStore, and executor.EventSink over a *statedb.DB.
func NewScheduler ¶
NewScheduler wraps a database handle.
func (*Scheduler) ClaimRequeue ¶
func (s *Scheduler) ClaimRequeue(sessionID, claimToken string, maxAttempts int, at time.Time) (executor.Session, bool, error)
ClaimRequeue is the exactly-once failover latch.
It mints a fresh claim token and hands it to a single conditional UPDATE. The supervisor that wins gets the session back carrying the new token; every other caller — a concurrent supervisor, or this one after a restart replaying the same unreachable transition — presents a token that no longer matches and receives executor.ErrSessionClaimLost.
The same UPDATE applies the failover cap (Task 20391): a session already re-dispatched maxAttempts times is closed as failover_exhausted instead of requeued, and exhausted reports it. The winner learns the session's fate from the row the claim wrote, not from a check of its own, so it cannot disagree with what was recorded.
The sentinel is translated at this boundary for the same reason store.go translates the enrollment sentinels: the supervisor distinguishes "I lost the race, do nothing" from "the database is broken, complain", and collapsing both into an opaque storage error would turn a benign race into log noise that hides a real outage.
func (*Scheduler) CloseSession ¶
CloseSession marks a session finished or failed.
func (*Scheduler) CountRunning ¶
CountRunning returns the number of in-flight sessions on an executor.
func (*Scheduler) ExecutorFailover ¶
func (s *Scheduler) ExecutorFailover(ev executor.FailoverEvent)
ExecutorFailover records one session moved off a dead node, including the case where no replacement was found, and — as executor.failover_exhausted — the case where the cap allowed none (Task 20391).
func (*Scheduler) ExecutorTransition ¶
func (s *Scheduler) ExecutorTransition(t executor.Transition)
ExecutorTransition records a state change in the audit log.
Failures are swallowed deliberately. The sink is called from probe goroutines, and an audit write that fails must not be able to stop the supervisor from *acting* on a node going unreachable — losing the log line is bad, losing the failover is worse.
func (*Scheduler) FailedOverProjects ¶ added in v0.0.5
FailedOverProjects returns the hub path of every project a session was ever failed over or exhausted for (Task 20391), sorted: the projects that can hold a task quarantined as a suspected node killer.
func (*Scheduler) ForgetHealth ¶
ForgetHealth drops an executor's health record, for use when an executor is deleted. A stale record would otherwise resurrect as a phantom row in `cloop executor ls` naming a backend that no longer exists.
func (*Scheduler) ListHealth ¶
ListHealth returns every persisted health record, ordered by executor ID.
func (*Scheduler) LoadHealth ¶
LoadHealth returns the persisted health for an executor. An executor that has never been probed yields the zero Health and a nil error: "not yet observed" is a normal state for a backend registered from config a millisecond ago, not a lookup failure the supervisor should log.
func (*Scheduler) LossOfHandle ¶ added in v0.0.5
func (s *Scheduler) LossOfHandle(executorID, handleID string) (HandleLoss, error)
LossOfHandle reads what the store recorded about the workload handleID on executorID. A workload with no session yields the zero HandleLoss and a nil error: "never tracked" is a normal answer, not a failed lookup.
func (*Scheduler) LostNodes ¶ added in v0.0.5
LostNodes returns every node the run behind sessionID has been lost on so far, oldest first: one entry per session in its failover chain that a claim took off an unreachable executor — requeued or failover_exhausted — with when the claim found it gone (Task 20391).
func (*Scheduler) OpenRequeuedSession ¶
OpenRequeuedSession records the replacement session created by a failover, linking it back to the session it replaces.
func (*Scheduler) OpenSession ¶
OpenSession records a dispatched workload as in flight.
The caller supplies the session ID and claim token (see NewSessionID and NewClaimToken) because it needs the token in hand before the workload starts: a session that becomes visible to a supervisor before its owner knows the token is a session the owner can no longer defend against requeue.
func (*Scheduler) RunningSessions ¶
RunningSessions returns the in-flight sessions on an executor. Passing "" returns every running session.
func (*Scheduler) RunningTasksOf ¶ added in v0.0.5
RunningTasksOf returns the tasks a session last recorded as running.
func (*Scheduler) SaveHealth ¶
SaveHealth persists a health record.
func (*Scheduler) SessionState ¶ added in v0.0.5
SessionState returns a session's recorded state and whether a failover has already opened a replacement for it — the two facts a re-dispatch checks before starting anything (Task 20391).
func (*Scheduler) SetRunningTasks ¶ added in v0.0.5
SetRunningTasks records which tasks a running session's workload is working on (Task 20391). It reports whether the session was still running to record it on.
type Store ¶
type Store struct {
// contains filtered or unexported fields
}
Store implements remote.Store over a *statedb.DB.
func (*Store) GetAgent ¶
func (s *Store) GetAgent(agentID string) (remote.AgentRecord, error)
GetAgent loads an agent by ID.
func (*Store) GetEnrollment ¶
func (s *Store) GetEnrollment(id string) (remote.EnrollmentRecord, error)
GetEnrollment loads a token by ID.
func (*Store) ListAgents ¶
func (s *Store) ListAgents() ([]remote.AgentRecord, error)
ListAgents returns every enrolled agent.
func (*Store) ListEnrollments ¶
func (s *Store) ListEnrollments() ([]remote.EnrollmentRecord, error)
ListEnrollments returns every token, newest first.
func (*Store) PutAgent ¶
func (s *Store) PutAgent(rec remote.AgentRecord) error
PutAgent persists an agent credential record.
func (*Store) PutEnrollment ¶
func (s *Store) PutEnrollment(rec remote.EnrollmentRecord) error
PutEnrollment persists a minted token record.
func (*Store) RedeemEnrollment ¶
RedeemEnrollment atomically claims a token.
func (*Store) RevokeAgent ¶
RevokeAgent marks an agent's credential revoked.
func (*Store) RevokeEnrollment ¶
RevokeEnrollment marks a token revoked.
Source Files