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 NewClaimToken() (string, error)
- func NewSessionID() (string, error)
- type Handles
- type Scheduler
- func (s *Scheduler) ClaimRequeue(sessionID, claimToken string, at time.Time) (executor.Session, 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) ForgetHealth(executorID string) error
- func (s *Scheduler) ListHealth() ([]executor.Health, error)
- func (s *Scheduler) LoadHealth(executorID string) (executor.Health, 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) SaveHealth(h executor.Health) 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 ( AuditEntityExecutor = "executor" AuditEventStateChange = "executor.state_change" AuditEventFailover = "executor.failover" 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 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 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) DeleteHandle ¶
DeleteHandle implements executor.HandleStore.
func (*Handles) ListHandles ¶
func (h *Handles) ListHandles(executorID string) ([]executor.HandleRecord, error)
ListHandles implements executor.HandleStore.
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, at time.Time) (executor.Session, 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 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.
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) 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) 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) SaveHealth ¶
SaveHealth persists a health record.
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