coordinator

package
v0.1.0 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Aug 18, 2026 License: AGPL-3.0 Imports: 24 Imported by: 0

Documentation

Overview

Package coordinator is LoadWave's control plane.

It holds the fleet together: agents join it, it decides how a run's load is divided between them, it merges everything they report into one coherent picture, and it serves that picture to the CLI and the dashboard. It is the only component that sees the whole run, which is why threshold evaluation and the pass/fail verdict live here rather than anywhere further down.

Index

Constants

View Source
const (
	DefaultAgentTimeout  = 15 * time.Second
	DefaultStartDelay    = 2 * time.Second
	DefaultMaxRunHistory = 50
)

Defaults for a zero Config.

View Source
const (
	PhasePending   = "pending"
	PhaseStarting  = "starting"
	PhaseRunning   = "running"
	PhaseStopping  = "stopping"
	PhaseCompleted = "completed"
	PhaseFailed    = "failed"
	PhaseAborted   = "aborted"
)

Phase names as they appear in the API, chosen to read well in a UI rather than to mirror the protobuf enum's spelling.

View Source
const (
	UpdateTick  = "tick"
	UpdateEvent = "event"
)

Update types pushed to live subscribers.

Variables

This section is empty.

Functions

func AnyBreached

func AnyBreached(results []ThresholdResult) bool

AnyBreached reports whether any evaluated threshold failed.

func IsTerminal

func IsTerminal(phase loadwavev1.RunPhase) bool

IsTerminal reports whether a phase means the run is over.

func NewRunID

func NewRunID(now time.Time) string

NewRunID builds a sortable, human-legible run identifier.

The timestamp prefix makes runs sort chronologically in a listing and in a directory of result files; the random suffix keeps two runs started in the same second apart.

func PhaseName

func PhaseName(phase loadwavev1.RunPhase) string

PhaseName renders a phase for the API.

Types

type AgentInfo

type AgentInfo struct {
	ID             string            `json:"id"`
	Hostname       string            `json:"hostname"`
	Version        string            `json:"version"`
	Cores          uint32            `json:"cores"`
	MaxWorkers     uint32            `json:"maxWorkers"`
	MaxVUs         uint32            `json:"maxVUs"`
	Labels         map[string]string `json:"labels,omitempty"`
	RemoteAddr     string            `json:"remoteAddr"`
	JoinedAt       time.Time         `json:"joinedAt"`
	LastSeen       time.Time         `json:"lastSeen"`
	ActiveVUs      uint32            `json:"activeVUs"`
	HealthyWorkers uint32            `json:"healthyWorkers"`
	Healthy        bool              `json:"healthy"`
	VUQuota        int               `json:"vuQuota"`

	// CPUPercent and MemBytes describe the agent process itself — its
	// supervisory footprint, not the workers it spawns. Zero until its
	// first heartbeat arrives.
	CPUPercent float64 `json:"cpuPercent"`
	MemBytes   uint64  `json:"memBytes"`

	// Workers is the per-process breakdown within this agent. Never nil:
	// this is JSON-encoded straight to the dashboard, which maps over it
	// unconditionally.
	Workers []WorkerInfo `json:"workers"`
}

AgentInfo is the operator-facing view of a connected agent.

type Config

type Config struct {
	// ListenAddr is where agents connect. ":0" picks a free port, which is
	// what the single-process `loadwave run` path uses.
	ListenAddr string

	Logger *slog.Logger

	// HeartbeatInterval is how often agents should report liveness.
	HeartbeatInterval time.Duration

	// MetricsInterval is the reporting period, and therefore the width of a
	// chart bucket.
	MetricsInterval time.Duration

	// AgentTimeout is how long an agent may go unheard from before it is
	// treated as gone. Generous relative to the heartbeat, so that one
	// dropped packet does not evict a healthy agent mid-run.
	AgentTimeout time.Duration

	// StartDelay is the lead time built into a run's agreed start instant, so
	// that every agent has received its orders before the clock starts.
	StartDelay time.Duration

	// Store configures each run's metric store.
	Store metrics.StoreConfig

	// MaxRunHistory is how many finished runs are retained in memory.
	MaxRunHistory int
}

Config describes a coordinator.

type Coordinator

type Coordinator struct {
	// contains filtered or unexported fields
}

Coordinator is the control plane.

func New

func New(cfg Config) (*Coordinator, error)

New prepares a coordinator. It does not listen; Run does that.

func (*Coordinator) ActiveRun

func (c *Coordinator) ActiveRun() *Run

ActiveRun returns the run in progress, or nil.

func (*Coordinator) Addr

func (c *Coordinator) Addr() string

Addr returns the address agents should dial, once the listener is up.

func (*Coordinator) Agents

func (c *Coordinator) Agents() []AgentInfo

Agents lists connected agents, newest first.

func (*Coordinator) GlobalEvents

func (c *Coordinator) GlobalEvents() []Event

GlobalEvents returns cluster-level events not attached to a run.

Never nil, even when empty: this feeds Snapshot.Events, which is JSON-encoded straight into the API, and a nil slice there marshals to `null` rather than `[]`.

func (*Coordinator) Lookup

func (c *Coordinator) Lookup(runID string) (*Run, bool)

Lookup returns a run by id.

func (*Coordinator) OnHeartbeat

func (c *Coordinator) OnHeartbeat(session *control.Session, beat *loadwavev1.NodeHeartbeat)

OnHeartbeat implements control.SessionHandler.

func (*Coordinator) OnJoin

func (c *Coordinator) OnJoin(_ context.Context, session *control.Session) error

OnJoin implements control.SessionHandler.

func (*Coordinator) OnLeave

func (c *Coordinator) OnLeave(session *control.Session)

OnLeave implements control.SessionHandler.

func (*Coordinator) OnLog

func (c *Coordinator) OnLog(session *control.Session, event *loadwavev1.LogEvent)

OnLog implements control.SessionHandler.

func (*Coordinator) OnMetrics

func (c *Coordinator) OnMetrics(_ *control.Session, batch *loadwavev1.MetricBatch)

OnMetrics implements control.SessionHandler.

func (*Coordinator) OnRunStatus

func (c *Coordinator) OnRunStatus(session *control.Session, update *loadwavev1.RunStatusUpdate)

OnRunStatus implements control.SessionHandler.

func (*Coordinator) Run

func (c *Coordinator) Run(ctx context.Context) error

Run listens for agents and serves until ctx is cancelled.

func (*Coordinator) RunSnapshot

func (c *Coordinator) RunSnapshot(runID string) (Snapshot, bool)

RunSnapshot renders one run's full state, live or finished.

func (*Coordinator) Runs

func (c *Coordinator) Runs() []Summary

Runs lists known runs, newest first.

func (*Coordinator) ScaleRun

func (c *Coordinator) ScaleRun(runID string, peakVUs int, ramp time.Duration) error

ScaleRun changes a running test's peak virtual user count.

A non-zero ramp introduces the change gradually. Spawning several hundred virtual users in a single tick measures how the service copes with a thundering herd, which is a different question from how it copes with the load level being asked for; the ramp is how an operator asks the second question.

func (*Coordinator) Snapshot

func (c *Coordinator) Snapshot() Snapshot

Snapshot renders the coordinator's whole state.

func (*Coordinator) StartRun

func (c *Coordinator) StartRun(cfg *scenario.Config, sourcePath string) (*Run, error)

StartRun distributes a test plan across the connected agents.

Only one run is permitted at a time. Concurrent runs would have to share the same worker processes and the same network capacity, and the resulting numbers would measure the interference rather than the system under test. sourcePath is the configuration file cfg was loaded from, or empty when there isn't one — a Go-defined scenario, a quick-check built from flags, or a configuration submitted to the dashboard directly.

func (*Coordinator) StopRun

func (c *Coordinator) StopRun(runID string, graceful bool, reason string) error

StopRun ends a run.

func (*Coordinator) Subscribe

func (c *Coordinator) Subscribe() *Subscription

Subscribe opens a live feed.

type EndpointTick

type EndpointTick struct {
	Avg       float64 `json:"avg"`
	Requests  uint64  `json:"requests"`
	ErrorRate float64 `json:"errorRate"`
}

EndpointTick is one endpoint's slice of a time bucket.

Average only, no percentiles: keeping a histogram per endpoint per second is what the store deliberately does not do. Whole-run percentiles per endpoint are in Snapshot.Endpoints.

type Event

type Event struct {
	Time    time.Time         `json:"time"`
	Level   string            `json:"level"`
	Source  string            `json:"source"`
	Message string            `json:"message"`
	Fields  map[string]string `json:"fields,omitempty"`
}

Event is something worth telling the operator about.

type Run

type Run struct {
	// contains filtered or unexported fields
}

Run is the coordinator's record of one execution.

Safe for concurrent use: the API serves it while control-stream goroutines update it.

func (*Run) Active

func (r *Run) Active() bool

Active reports whether the run is still going.

func (*Run) Breached

func (r *Run) Breached() bool

Breached reports whether any threshold has failed at any point.

It latches deliberately. A p95 that recovers by the end of the run still breached, and a CI gate that only looked at the final instant would let a genuine regression through.

func (*Run) Events

func (r *Run) Events() []Event

Events returns a copy of the run's event log.

Never nil, even when empty: this is JSON-encoded straight into the API, and a nil slice there marshals to `null` rather than `[]` — which is exactly the kind of surprise a frontend array method doesn't guard against.

func (*Run) GraceBudget

func (r *Run) GraceBudget() time.Duration

GraceBudget is how long the plan allows in-flight iterations to finish.

func (*Run) ID

func (r *Run) ID() string

ID returns the run's identifier.

func (*Run) Participants

func (r *Run) Participants() []participant

Participants returns a copy of the agent roster.

func (*Run) Phase

func (r *Run) Phase() loadwavev1.RunPhase

Phase returns the current phase.

func (*Run) Plan

func (r *Run) Plan() *loadwavev1.TestPlan

Plan returns the plan being executed.

func (*Run) SourcePath

func (r *Run) SourcePath() string

SourcePath is the configuration file this run was started from, or empty when there isn't one.

func (*Run) StoppingFor

func (r *Run) StoppingFor() time.Duration

StoppingFor reports how long the run has been trying to stop, or zero if it has not been asked to.

func (*Run) Store

func (r *Run) Store() *metrics.Store

Store returns the run's metric store.

func (*Run) Summary

func (r *Run) Summary(profile string) Summary

Summary renders the run for the API.

func (*Run) Thresholds

func (r *Run) Thresholds() []ThresholdResult

Thresholds returns the latest evaluation.

Never nil, even before the first evaluation: this is JSON-encoded straight into the API, and a nil slice there marshals to `null` rather than `[]`.

type ScenarioTick

type ScenarioTick struct {
	VUs        float64 `json:"vus"`
	Iterations uint64  `json:"iterations"`
	Requests   uint64  `json:"requests"`
	ErrorRate  float64 `json:"errorRate"`
	P95        float64 `json:"p95"`
}

ScenarioTick is one scenario's slice of a time bucket.

type Snapshot

type Snapshot struct {
	Build      buildinfo.Info          `json:"build"`
	Run        *Summary                `json:"run,omitempty"`
	Runs       []Summary               `json:"runs"`
	Agents     []AgentInfo             `json:"agents"`
	Ticks      []TickDTO               `json:"ticks"`
	Series     []metrics.SeriesSummary `json:"series"`
	Events     []Event                 `json:"events"`
	Resolution float64                 `json:"resolutionSeconds"`

	// Totals holds one correctly merged aggregate per metric. Clients must
	// use these for whole-run figures rather than folding Series, which
	// cannot be done correctly outside the store.
	Totals map[string]metrics.SeriesSummary `json:"totals,omitempty"`

	// Endpoints is the per-request-name breakdown, with percentiles
	// recomputed from each endpoint's merged distribution.
	Endpoints []metrics.EndpointSummary `json:"endpoints,omitempty"`

	// Failures explains what went wrong, which the metrics only count.
	Failures []metrics.FailureSummary `json:"failures,omitempty"`
}

Snapshot is the complete current state, served to a client on connect.

type Subscription

type Subscription struct {
	// contains filtered or unexported fields
}

Subscription is one client's live feed.

func (*Subscription) Close

func (s *Subscription) Close()

Close ends the subscription. It is safe to call more than once.

func (*Subscription) Done

func (s *Subscription) Done() <-chan struct{}

Done is closed when the subscription ends.

func (*Subscription) Updates

func (s *Subscription) Updates() <-chan Update

Updates returns the channel of live updates. It is never closed; select on Done alongside it.

type Summary

type Summary struct {
	ID         string            `json:"id"`
	Name       string            `json:"name"`
	Phase      string            `json:"phase"`
	CreatedAt  time.Time         `json:"createdAt"`
	StartAt    time.Time         `json:"startAt,omitempty"`
	StartedAt  time.Time         `json:"startedAt,omitempty"`
	EndedAt    time.Time         `json:"endedAt,omitempty"`
	ElapsedSec float64           `json:"elapsedSeconds"`
	PeakVUs    int               `json:"peakVUs"`
	Profile    string            `json:"profile"`
	BaseURL    string            `json:"baseURL"`
	StopReason string            `json:"stopReason,omitempty"`
	Failure    string            `json:"failure,omitempty"`
	Breached   bool              `json:"thresholdsBreached"`
	Tags       map[string]string `json:"tags,omitempty"`

	Participants []participant     `json:"participants"`
	Thresholds   []ThresholdResult `json:"thresholds"`
	Stats        metrics.Stats     `json:"stats"`
}

Summary is the API's view of a run.

type ThresholdResult

type ThresholdResult struct {
	Metric string  `json:"metric"`
	Stat   string  `json:"stat"`
	Op     string  `json:"op"`
	Value  float64 `json:"value"`

	// Actual is the observed statistic. Meaningful only when Evaluated.
	Actual float64 `json:"actual"`

	// Evaluated is false when the metric has not been observed yet. A
	// threshold on a metric that never appeared is reported as unevaluated
	// rather than as a pass, because "we never measured it" and "it was fine"
	// are very different things to hand back to a CI pipeline.
	Evaluated bool `json:"evaluated"`

	Passed bool `json:"passed"`

	// AbortOnFail means a breach should end the run immediately.
	AbortOnFail bool `json:"abortOnFail"`

	// Description renders the assertion for display, e.g.
	// "http_req_duration p95 < 500".
	Description string `json:"description"`
}

ThresholdResult is one threshold's verdict.

func AbortRequested

func AbortRequested(results []ThresholdResult) (ThresholdResult, bool)

AbortRequested reports whether a failing threshold asked for the run to be cut short.

func EvaluateThresholds

func EvaluateThresholds(store *metrics.Store, thresholds []*loadwavev1.Threshold) []ThresholdResult

EvaluateThresholds checks every threshold against the run's cumulative metrics.

Evaluation is cumulative rather than windowed: a threshold answers "was this run acceptable overall", which is the question a CI gate is asking. A momentary spike that the run recovered from does not fail the build unless it moved the whole-run statistic past the line.

type TickDTO

type TickDTO struct {
	T          int64   `json:"t"`
	VUs        uint32  `json:"vus"`
	Requests   uint64  `json:"requests"`
	Failures   uint64  `json:"failures"`
	Iterations uint64  `json:"iterations"`
	RPS        float64 `json:"rps"`
	ErrorRate  float64 `json:"errorRate"`
	Avg        float64 `json:"avg"`
	P50        float64 `json:"p50"`
	P90        float64 `json:"p90"`
	P95        float64 `json:"p95"`
	P99        float64 `json:"p99"`

	Status    map[string]uint64       `json:"status,omitempty"`
	Scenarios map[string]ScenarioTick `json:"scenarios,omitempty"`

	// Endpoints is the per-request-name breakdown, which is what the response
	// time chart plots one line from.
	Endpoints map[string]EndpointTick `json:"endpoints,omitempty"`
}

TickDTO is one second of the run, shaped for charting.

The wire format is deliberately flat and pre-computed rather than a generic metric dump. The dashboard redraws several series many times a second, and having the browser derive rates and dig percentiles out of nested maps on every frame is exactly the sort of work that makes a live chart stutter.

type Update

type Update struct {
	Type       string            `json:"type"`
	Run        *Summary          `json:"run,omitempty"`
	Agents     []AgentInfo       `json:"agents,omitempty"`
	Ticks      []TickDTO         `json:"ticks,omitempty"`
	Thresholds []ThresholdResult `json:"thresholds,omitempty"`
	Events     []Event           `json:"events,omitempty"`
}

Update is one message on the live stream.

type WorkerInfo

type WorkerInfo struct {
	ID         string  `json:"id"`
	Index      uint32  `json:"index"`
	ActiveVUs  uint32  `json:"activeVUs"`
	CPUPercent float64 `json:"cpuPercent"`
	MemBytes   uint64  `json:"memBytes"`
}

WorkerInfo is one worker process's resource usage, as its agent reported it — the detail an agent-level aggregate would hide, such as one process starved for CPU while its siblings on the same host are not.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL