controller

type Controller ¶

type Controller struct {
	Org     func() (*policy.Org, error) // e.g. policy.Load of the served policy dir
	Metrics promote.MetricSource
	Writer  Writer     // nil: no PRs (logged)
	Kill    KillSwitch // nil: rollbacks are not enforced until the PR merges (logged, notified)
	// KillCaveat, when set, is why a successful Kill is not known to be
	// enforced by gateways (e.g. `halo controller run` without
	// --killswitch-served); messages report it instead of claiming enforcement.
	KillCaveat string
	// Kills is read each tick: a killed running experiment is not evaluated
	// (its evidence is all-control) and humans are told once. nil = no check.
	Kills        *KillStore
	Notifier     Notifier // nil: none
	State        *StateLog
	VerdictsPath string // "" = do not persist verdicts
	Clock        func() time.Time
	Log          *slog.Logger
	TickTimeout  time.Duration // default 2m; bounds one whole tick

	// RolloutStateDir holds one hash-chained state file per Rollout ("" =
	// rollouts are not driven); PolicyDir is where step scorecards are read.
	RolloutStateDir, PolicyDir string
	// RolloutStates, when set, replaces RolloutStateDir (the shared store of an HA deployment).
	RolloutStates rollout.StateStore
	// VerdictDocs, when set, replaces VerdictsPath: one VerdictDocKind
	// document per experiment, which every replica's console reads (HA).
	VerdictDocs rollout.Docs
	// Governance, when set, returns halo-server's incident freeze and the
	// approvals recorded for a rollout's live step (ADR-0030). nil: policy
	// freeze windows only, and a step an approval rule covers waits (its
	// approvals live on halo-server).
	Governance func(ctx context.Context, org *policy.Org, r *policy.Rollout) (rollout.Governance, error)
	// contains filtered or unexported fields
}

Controller wires the loop. Org, Metrics and State are required.

func (c *Controller) Run(ctx context.Context, interval time.Duration)

func (c *Controller) Tick(ctx context.Context) error

func (c *Controller) WriteMetrics(w io.Writer)

type Event ¶

type Event struct {
	// ID is the delivery's idempotency key: the same for every attempt at one
	// notification, by any controller replica, so a receiver can drop a
	// duplicate sent across a leader failover (webhook: X-Halo-Event-Id).
	ID string `json:"id,omitempty"`
	// Rollout is set for rollout events; Experiment is then its backing experiment.
	Rollout     string          `json:"rollout,omitempty"`
	Experiment  string          `json:"experiment"`
	Axis        string          `json:"axis,omitempty"`
	Verdict     promote.Verdict `json:"verdict"`
	Reason      string          `json:"reason"`
	Killed      bool            `json:"killed"` // traffic-plane kill switch is active for the experiment
	KillOutcome KillOutcome     `json:"killOutcome,omitempty"`
	KillError   string          `json:"killError,omitempty"` // detail for every non-enforced outcome
	PRURL       string          `json:"prURL,omitempty"`
	PRError     string          `json:"prError,omitempty"`
	Report      promote.Report  `json:"report"`
	At          time.Time       `json:"at"`
	// Resolved marks the all-clear for an earlier event of the same Verdict:
	// the condition that raised it is over (a human merged the pause or
	// advanced the rollout, or the kill was lifted). Only sinks that opt in
	// with the "resolved" event kind receive it (webhook, PagerDuty).
	Resolved bool `json:"resolved,omitempty"`
}

Event is what the controller tells humans about a verdict it acted on.

func (e Event) DedupKey() string

func (e Event) Kind() string

func (e Event) PagerDutySeverity() string

critical  a rollback whose exposure continues: the kill switch is not
          enforced (failed, not configured, recorded only, or the
          evidence was not gateway-sourced), so users stay on the
          treatment until a human merges the pause PR
error     a rollback whose kill switch is enforced (mitigated, needs
          follow-up)
warning   a rollout paused on a gate breach, or a killed experiment
          held until someone unkills it
info      anything else (only delivered when an operator opts it in)

func (e Event) Text() string

type EventFilter ¶ added in v0.5.0

type EventFilter map[string]bool

EventFilter is the set of kinds a sink delivers. nil means every event except resolved (what Slack and the webhook always received); a non-nil set is exact. A resolved event needs "resolved" in the set.

func ParseEventFilter(flag, spec string, allowResolved bool) (EventFilter, error)

func (f EventFilter) Allows(e Event) bool

type FlagSet ¶ added in v0.5.0

type FlagSet interface {
	StringVar(p *string, name, value, usage string)
}

FlagSet is the part of flag.FlagSet and pflag.FlagSet NotifyFlags needs, so halo-server and `halo controller run` register identical flags.

type GitWriter ¶

type GitWriter struct {
	RepoDir string // git checkout (or any dir inside one)
	Subdir  string // policy root inside RepoDir
	Base    string // PR base branch; "" = repo default
	Opener  promote.PROpener
}

GitWriter implements Writer with promote.PlanStatus and a promote.PROpener (GHOpener: branch in a temp worktree, git push, gh pr create). It is the one "set experiment status + open PR" implementation: `halo controller run` uses it on the CI checkout; halo-server's server.GitPolicyWriter uses it on its dedicated clone after syncing that clone to origin (under its own lock). It never touches RepoDir's branch, index or working tree.

func (w *GitWriter) FindPR(ctx context.Context, marker string) (string, bool, error)

func (w *GitWriter) ProposeRollout(ctx context.Context, plan func(string) (*promote.Change, error), branch, title string, body func(string) string) (string, error)

func (w *GitWriter) ProposeStatus(ctx context.Context, experiment, status, title, body string) (string, error)

type KillFunc ¶

type KillFunc func(ctx context.Context, experiment, by, reason string) (bool, error)

KillFunc adapts a function (e.g. server.KillExperiment, which also audit-logs) to KillSwitch.

func (f KillFunc) Kill(ctx context.Context, experiment, by, reason string) (bool, error)

type KillOutcome ¶

type KillOutcome string

KillOutcome is the typed result of the kill-switch step of a rollback.

const (
	KillEnforced      KillOutcome = "enforced"             // tripped and enforced by gateways
	KillNotGateway    KillOutcome = "not_gateway_evidence" // evidence not gateway-sourced; human decides
	KillNotConfigured KillOutcome = "not_configured"       // no kill switch wired
	KillRecorded      KillOutcome = "recorded_only"        // written, enforcement not guaranteed (KillCaveat)
	KillFailed        KillOutcome = "failed"               // the kill call errored
)

type KillRecord ¶

type KillRecord struct {
	Experiment string    `json:"experiment"`
	Killed     bool      `json:"killed"`
	By         string    `json:"by"`
	Reason     string    `json:"reason,omitempty"`
	At         time.Time `json:"at"`
}

KillRecord is one line of killswitch.jsonl; the last line per experiment wins.

type KillStore ¶

type KillStore struct {
	Now func() time.Time
	// contains filtered or unexported fields
}

KillStore is the persisted kill state halo-server serves to gateways: an append-only JSONL log (killswitch.jsonl in the data dir) replayed into memory. Appends from another process sharing the file (`halo controller run` against halo-server's data dir) are picked up on the next read. ponytail: whole-file re-read when the size changes; kills are rare. Cross-process appends rely on O_APPEND single-line writes (local filesystem).

NewKillStore puts it on a shared log instead (HA): every replica then catches up with the others' kills on each read.

func NewKillStore(ctx context.Context, o storage.Opener) (*KillStore, error)

func OpenKillStore(dir string) (*KillStore, error)

func (s *KillStore) Close() error

func (s *KillStore) Freeze(ctx context.Context, by, reason string) (bool, error)

func (s *KillStore) Incident() (*KillRecord, error)

func (s *KillStore) Kill(ctx context.Context, experiment, by, reason string) (bool, error)

func (s *KillStore) Killed() ([]KillRecord, uint64, error)

func (s *KillStore) Unfreeze(ctx context.Context, by, reason string) (bool, error)

func (s *KillStore) Unkill(ctx context.Context, experiment, by, reason string) (bool, error)

type KillSwitch ¶

type KillSwitch interface {
	// Kill marks experiment killed; changed is false when it already was.
	Kill(ctx context.Context, experiment, by, reason string) (changed bool, err error)
}

KillSwitch trips the traffic-plane kill switch for an experiment.

type Named ¶

type Named interface {
	Name() string
}

Named is implemented by notifiers with a stable channel name. The controller records each channel's delivery separately (ActionNotify + ":" + name), so a failing webhook is retried without re-posting to Slack.

type Notifier ¶

type Notifier interface {
	Notify(ctx context.Context, e Event) error
}

Notifier delivers events. Implementations must bound their own latency.

type Notifiers ¶

type Notifiers []Notifier

Notifiers fans out to every notifier and joins their errors.

func NotifiersFromFiles(slackURLFile, webhookURLFile, webhookSecretFile string) (Notifiers, error)

func (ns Notifiers) Notify(ctx context.Context, e Event) error

func (ns Notifiers) WriteMetrics(w io.Writer)

type NotifyConfig ¶ added in v0.5.0

type NotifyConfig struct {
	SlackURLFile, SlackEvents string
	TeamsURLFile, TeamsEvents string

	WebhookURLFile, WebhookSecretFile, WebhookEvents string

	PagerDutyRoutingKeyFile, PagerDutyEvents string
	PagerDutyURL                             string // "" = PagerDutyEventsURL (EU region, or a test double)

	Retry RetryPolicy // applied to every sink
}

NotifyConfig is the notification setup shared by `halo controller run` and `halo-server --controller`. Every secret (a webhook URL embeds a token; a routing key is one) is read from a file, never from a flag value, and no error echoes one. A sink is on when its URL/key file is set.

func (c NotifyConfig) Build() (Notifiers, error)

type PRFinder ¶ added in v0.3.0

type PRFinder interface {
	FindPR(ctx context.Context, marker string) (url string, found bool, err error)
}

PRFinder is implemented by PR writers that can find a pull request they opened by a marker embedded in its body (GitWriter, through a promote.PRFinder opener). The controller uses it when an intent shows a PR may already have been opened by a deposed leader.

type PagerDuty ¶ added in v0.5.0

type PagerDuty struct {
	RoutingKey string
	URL        string // default PagerDutyEventsURL
	Source     string // payload.source, the affected system (default "halos-controller")
	Client     *http.Client
	Retry      RetryPolicy // zero: one attempt
}

PagerDuty raises and clears incidents through the Events API v2.

A paging event is a "trigger" with a stable dedup_key, so a repeat for the same rollout or experiment folds into one incident; the all-clear (the condition ended: a human merged the pause or advanced the rollout, or the kill was lifted) is a "resolve" with the same dedup_key. Informational events (a PR opened, a promotion proposed, a freeze) do not page: the default event filter excludes them, and they map to severity "info" if an operator opts them in.

The routing key is a secret: it travels only in the request body, and no error or log line contains it.

func (PagerDuty) Name() string

func (p PagerDuty) Notify(ctx context.Context, e Event) error

type RetryPolicy ¶ added in v0.5.0

type RetryPolicy struct {
	Attempts int           // total tries including the first; <= 1: no retry
	Base     time.Duration // wait before the second try; doubles each retry
	MaxWait  time.Duration // cap on one wait; a longer Retry-After ends the retrying
	Budget   time.Duration // cap on the whole delivery, waits included; 0: unbounded

	// Sleep waits d or until ctx ends (default: a real timer); Now is the
	// clock for Retry-After dates and Budget (default time.Now). Both are
	// injectable so tests need no real waiting.
	Sleep func(ctx context.Context, d time.Duration) error
	Now   func() time.Time
}

RetryPolicy bounds how hard one delivery tries before giving up. The zero value makes a single attempt (what a bare Slack{} or Webhook{} does). A delivery that still fails is not lost: the controller records a notification only once a channel accepted it, so the next tick tries again.

Retried: transport errors and timeouts, HTTP 429 and 5xx. Never retried: other 4xx (a bad URL, routing key or payload does not heal) and a cancelled context.

type RolloutWriter ¶

type RolloutWriter interface {
	ProposeRollout(ctx context.Context, plan func(dir string) (*promote.Change, error), branch, title string, body func(patch string) string) (prURL string, err error)
}

RolloutWriter opens a rollout PR: plan builds the change in the writer's checkout dir, body renders the PR body around its patch. GitWriter and server.GitPolicyWriter implement it. It must never merge.

type Sink ¶ added in v0.5.0

type Sink struct {
	SinkName string
	Notifier Notifier
	Events   EventFilter
	// contains filtered or unexported fields
}

Sink is one named delivery channel: a Notifier behind an event filter, with delivery counters. The name is the channel key in the controller's state (ActionNotify + ":" + name), so a failing sink is retried without re-posting to the others.

func (s *Sink) Name() string

func (s *Sink) Notify(ctx context.Context, e Event) error

func (s *Sink) ResolvesFor(trigger Event) bool

type Slack ¶

type Slack struct {
	URL    string
	Client *http.Client
	Retry  RetryPolicy // zero: one attempt
}

Slack posts to a Slack(-compatible) incoming webhook.

func (Slack) Name() string

func (s Slack) Notify(ctx context.Context, e Event) error

type StateLog ¶

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

StateLog remembers which (experiment, verdict, action) the controller has already done, so repeated ticks never open a second PR or send a second notification. An experiment's entries are forgotten once it is seen not running (a human merged the pause/conclude), so a later restart starts clean. ponytail: no compaction; one line per action per experiment run.

func OpenStateLog(dir string) (*StateLog, error)

func OpenStateLogOn(ctx context.Context, o storage.Opener) (*StateLog, error)

func (l *StateLog) Close() error

func (l *StateLog) Done(exp, verdict, action string) (string, bool)

func (l *StateLog) Names() []string

func (l *StateLog) Record(exp, verdict, action, detail string, at time.Time) error

func (l *StateLog) Reset(exp string, at time.Time) error

type Teams ¶ added in v0.5.0

type Teams struct {
	URL    string
	Client *http.Client
	Retry  RetryPolicy // zero: one attempt
}

Teams posts an Adaptive Card to a Microsoft Teams Workflows (Power Automate) webhook: the "When a Teams webhook request is received" trigger (the "Send webhook alerts to a channel" template). It expects a {"type":"message","attachments":[{"contentType": "application/vnd.microsoft.card.adaptive","content":<card>}]} body, which is what this sends.

This is deliberately not the legacy Office 365 connector ("Incoming Webhook", MessageCard JSON): Microsoft retired Office 365 connectors in Teams, and a legacy URL no longer delivers. NewTeams refuses those hosts so the failure is loud at startup instead of silent at 3 a.m.

func (Teams) Name() string

func (t Teams) Notify(ctx context.Context, e Event) error

type Webhook ¶

type Webhook struct {
	URL    string
	Secret []byte
	Client *http.Client
	Now    func() time.Time // default time.Now
	Retry  RetryPolicy      // zero: one attempt
}

Webhook posts the Event JSON, signed with Secret when set.

func (Webhook) Name() string

func (w Webhook) Notify(ctx context.Context, e Event) error

type Writer ¶

type Writer interface {
	ProposeStatus(ctx context.Context, experiment, status, title, body string) (prURL string, err error)
}

Writer opens a policy-repo PR setting an experiment's status. It must never merge.