observ

package
v0.4.0-beta.5 Latest Latest
Warning

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

Go to latest
Published: Aug 22, 2026 License: MIT Imports: 32 Imported by: 0

Documentation

Overview

Package observ is the FlatRun observability engine: it collects per-container metrics, keeps a bounded recent-history window for the native UI, stores the longer history, and hands the same data to anything else two ways: pushed over OTLP to a configured backend, or scraped in Prometheus format. Metric names follow the OpenTelemetry container semantic conventions.

Index

Constants

View Source
const (
	ComparisonAbove = "above"
	ComparisonBelow = "below"
)

Comparison is how a rule's threshold is read.

View Source
const (
	AlertOK      = "ok"
	AlertPending = "pending"
	AlertFiring  = "firing"
)

AlertState is where a rule stands.

View Source
const (
	ActionNone    = ""
	ActionRestart = "restart"
)

Alert actions.

View Source
const (
	HealthHealthy   = "healthy"
	HealthUnhealthy = "unhealthy"
	HealthStarting  = "starting"
	HealthNone      = "none"
)

Health status values as reported by Docker's HEALTHCHECK.

View Source
const (
	LogLevelDebug = "debug"
	LogLevelInfo  = "info"
	LogLevelWarn  = "warn"
	LogLevelError = "error"
	LogLevelFatal = "fatal"
)

Log levels a rule can gate on, ordered by severity.

View Source
const (
	MetricCPUUsage    = "container.cpu.usage"
	MetricMemoryUsage = "container.memory.usage"
	MetricMemoryLimit = "container.memory.limit"
	MetricNetworkRx   = "container.network.io.rx"
	MetricNetworkTx   = "container.network.io.tx"
)

OpenTelemetry container metric names (semconv). Emitting these verbatim keeps FlatRun's metrics interoperable with any OTel backend.

View Source
const (
	MetricHostCPU      = "system.cpu.utilization"
	MetricHostMemUtil  = "system.memory.utilization"
	MetricHostMemUsage = "system.memory.usage"
	MetricHostMemLimit = "system.memory.limit"
	MetricHostDisk     = "system.disk.utilization"
)

Host (system-wide) metric names, semconv system.* conventions. These answer "is the machine in trouble", which per-container percentages against a container's own limit cannot. Utilizations are percentages so an alert can target them directly; usage/limit are bytes for charting.

View Source
const HostContainer = "host"

HostContainer is the reserved Container value for host series; their Deployment is empty, so host series never collide with a real container.

View Source
const (
	ResponderNotify = "notify"
)

Variables

View Source
var ConfigSchema = map[string]any{
	"sample_interval_seconds":  map[string]any{"type": "number", "label": "Sample interval (seconds)", "default": 5, "min": 1},
	"auto_restart":             map[string]any{"type": "boolean", "label": "Auto-restart unhealthy containers", "default": true},
	"restart_cooldown_seconds": map[string]any{"type": "number", "label": "Restart cooldown (seconds)", "default": 120, "min": 10},
	"retention_days":           map[string]any{"type": "number", "label": "Keep history for (days)", "default": 7, "min": 1},
	"otlp_endpoint":            map[string]any{"type": "string", "label": "OTLP endpoint", "placeholder": "http://localhost:4318", "help": "Push metrics to an OpenTelemetry backend. Leave empty to only serve them for scraping."},
	"log_triage":               map[string]any{"type": "boolean", "label": "Let log rules ask the assistant", "default": false, "help": "Log rules that opt in can have the assistant explain an incident. Bounded by the agent's daily triage cap."},
	"triage_context_lines":     map[string]any{"type": "number", "label": "Lines of context per incident", "default": 12, "min": 1, "max": 40},
}

ConfigSchema describes the config for the settings form the UI renders.

View Source
var PluginInfo = pluginapi.Info{
	Name:         "observability",
	Version:      "0.1.0",
	DisplayName:  "Observability",
	Description:  "Per-deployment metrics and health, OpenTelemetry-native.",
	Capabilities: []string{"metrics", "docker"},
	ConfigSchema: ConfigSchema,
	UIExtensions: []pluginapi.UIExtension{
		{Slot: "deployment.detail", Kind: "metrics-panel", Title: "Metrics & Health", Icon: "activity", Endpoint: "/metrics/deployment"},
		{Slot: "settings", Kind: "form", Title: "Observability", Icon: "activity", Endpoint: "/config"},
	},
}

PluginInfo identifies the observability app to the host and declares the UI it contributes: a metrics + health panel inside each deployment, and a settings form for its config.

Functions

func DockerComposeRestart

func DockerComposeRestart(dir string) error

DockerComposeRestart restarts every service of a deployment by running compose in its directory. It restarts the deployment rather than a single container, which is what an operator means by "restart it".

func DockerRestart

func DockerRestart(container string) error

DockerRestart restarts a container by name.

func Handler

func Handler(store *Store, history *MetricsDB, health healthReporter, cfg configAccess, apply func(Config)) http.Handler

func HandlerWithAlerts

func HandlerWithAlerts(store *Store, history *MetricsDB, health healthReporter, cfg configAccess, apply func(Config), al alerts) http.Handler

HandlerWithAlerts is Handler plus the rule endpoints.

func KnownResponders

func KnownResponders() []string

func RegisterResponder

func RegisterResponder(r Responder)

RegisterResponder makes a responder available to rules, replacing one of the same name.

func RunPlugin

func RunPlugin() error

RunPlugin collects metrics, watches container health, restarts unhealthy containers, and serves it all until the host stops the process. It is the entry point for both the standalone plugin binary and the agent's self-exec subcommand.

func StartOTLPExport

func StartOTLPExport(ctx context.Context, store *Store, endpoint string) (func(context.Context) error, error)

StartOTLPExport pushes the collected metrics to an OTLP endpoint until ctx is done, and returns a shutdown that flushes what is pending.

The metrics are read from the same store the built-in UI draws, so an external backend and FlatRun's own views can never disagree about what a container did.

Values are reported as observable gauges read at export time rather than pushed on every sample, which is what lets the export interval differ from the sample interval without either side having to buffer.

Types

type ActionRunner

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

ActionRunner carries out a firing rule's action. It mirrors the health watcher's guardrails so a metric alert cannot turn into a restart loop: only FlatRun-managed deployments are touched, and a deployment is not restarted again until the cooldown has passed.

func NewActionRunner

func NewActionRunner(restart DeploymentRestartFunc, managed func(string) bool, cooldown time.Duration, dataDir string) *ActionRunner

func (*ActionRunner) Run

func (a *ActionRunner) Run(ev AlertEvent) string

Run executes the event's action and returns a short message describing what it did, or "" when it did nothing (not a restart action, no deployment, not managed, still cooling down, or the restart failed).

type AlertEngine

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

AlertEngine evaluates rules against the latest samples on a timer.

It reports transitions only. A rule that stays breached is one alert, not one every tick: an operator who is told the same thing every fifteen seconds stops reading any of it.

func NewAlertEngine

func NewAlertEngine(store *Store) *AlertEngine

func (*AlertEngine) Events

func (e *AlertEngine) Events() []AlertEvent

Events returns the recorded state changes, most recent last.

func (*AlertEngine) Firing

func (e *AlertEngine) Firing() []AlertEvent

Firing lists what is breached now, one entry per rule and container.

This is the state of things rather than a reading of the history: a rule that breaks, recovers and breaks again leaves a firing event behind each time, and reporting every one of them would list the same rule once per bad day it has had.

func (*AlertEngine) OnAction

func (e *AlertEngine) OnAction(fn func(AlertEvent))

OnAction registers the sink that carries out a firing rule's action.

func (*AlertEngine) OnAlert

func (e *AlertEngine) OnAlert(fn func(AlertEvent))

OnAlert registers the sink for state changes.

func (*AlertEngine) Rules

func (e *AlertEngine) Rules() []AlertRule

Rules returns the current rule set.

func (*AlertEngine) Run

func (e *AlertEngine) Run(stop <-chan struct{}, interval time.Duration)

Run evaluates on each tick until stopped.

func (*AlertEngine) SetRules

func (e *AlertEngine) SetRules(rules []AlertRule)

SetRules replaces the rule set, forgetting the state of rules that no longer exist.

type AlertEvent

type AlertEvent struct {
	RuleID     string    `json:"rule_id"`
	RuleName   string    `json:"rule_name"`
	Deployment string    `json:"deployment"`
	Container  string    `json:"container"`
	Metric     string    `json:"metric"`
	Value      float64   `json:"value"`
	Threshold  float64   `json:"threshold"`
	Comparison string    `json:"comparison"`
	State      string    `json:"state"`
	At         time.Time `json:"at"`
	// Targets and Action are copied from the rule so the event is self-contained
	// for the notification and action sinks.
	Targets []string `json:"targets,omitempty"`
	Action  string   `json:"action,omitempty"`
	// Snapshot is the top consuming containers at the moment a rule fired, so a
	// notification and the dashboard can show what was using the resource.
	Snapshot []Consumer `json:"snapshot,omitempty"`
}

AlertEvent is a rule changing state, which is the only thing worth telling anyone about.

func (AlertEvent) Message

func (e AlertEvent) Message() string

Message renders the event the way an operator reads it in a notification.

type AlertRule

type AlertRule struct {
	ID         string  `json:"id" yaml:"id"`
	Name       string  `json:"name" yaml:"name"`
	Deployment string  `json:"deployment,omitempty" yaml:"deployment,omitempty"`
	Metric     string  `json:"metric" yaml:"metric"`
	Comparison string  `json:"comparison" yaml:"comparison"`
	Threshold  float64 `json:"threshold" yaml:"threshold"`
	ForSeconds int     `json:"for_seconds" yaml:"for_seconds"`
	Enabled    bool    `json:"enabled" yaml:"enabled"`
	// Targets are the notification target ids this rule delivers to. Empty means
	// every configured target, which is the original behaviour.
	Targets []string `json:"targets,omitempty" yaml:"targets,omitempty"`
	// Action is an optional remediation taken when the rule fires. "" is notify
	// only; ActionRestart restarts the offending deployment.
	Action string `json:"action,omitempty" yaml:"action,omitempty"`
}

AlertRule fires when a metric stays past a threshold for long enough.

The duration is what separates an alert from a twitch: a container is briefly at 100% CPU every time it starts, and a rule without one would page an operator for it.

func (AlertRule) Validate

func (r AlertRule) Validate() error

Validate reports why a rule cannot be used, if it cannot.

type AlertStore

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

AlertStore keeps the rules in a flat file beside the rest of FlatRun's state, so they are readable and editable without the UI, like everything else here.

func NewAlertStore

func NewAlertStore(basePath string) *AlertStore

func (*AlertStore) Load

func (s *AlertStore) Load() []AlertRule

Load reads the rules, returning none when the file has never been written.

func (*AlertStore) Save

func (s *AlertStore) Save(rules []AlertRule) error

Save replaces the rules, giving any new one an id.

type Collector

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

Collector periodically reads a StatsSource and records the readings into a Store.

func NewCollector

func NewCollector(store *Store, source StatsSource, interval time.Duration) *Collector

func (*Collector) Run

func (c *Collector) Run(ctx context.Context)

Run collects on each tick until ctx is cancelled. A failing read is skipped, not fatal, so a transient Docker hiccup does not stop collection.

type Config

type Config struct {
	SampleIntervalSeconds  int  `yaml:"sample_interval_seconds" json:"sample_interval_seconds"`
	AutoRestart            bool `yaml:"auto_restart" json:"auto_restart"`
	RestartCooldownSeconds int  `yaml:"restart_cooldown_seconds" json:"restart_cooldown_seconds"`
	// RetentionDays bounds how far back stored history goes. Samples older than the recent
	// window are averaged into one point a minute, so a long retention is cheap.
	RetentionDays int `yaml:"retention_days" json:"retention_days"`
	// OTLPEndpoint is where metrics are pushed, if anywhere. An http or https URL speaks
	// OTLP/HTTP; a bare host:port speaks OTLP/gRPC. Left empty, the standard
	// OTEL_EXPORTER_OTLP_ENDPOINT environment variable is honoured instead, and with
	// neither set nothing is pushed and the metrics are still there to scrape.
	OTLPEndpoint string `yaml:"otlp_endpoint,omitempty" json:"otlp_endpoint,omitempty"`
	// Off unless turned on here, and still opt-in per rule after that.
	LogTriage bool `yaml:"log_triage" json:"log_triage"`
	// Bounds what an incident carries, and so the most a triage can be asked to read.
	TriageContextLines int `yaml:"triage_context_lines,omitempty" json:"triage_context_lines,omitempty"`
}

Config is the observability app's user-tunable settings, stored flat in .flatrun/observability.yml.

func DefaultConfig

func DefaultConfig() Config

DefaultConfig returns the built-in defaults.

type ConfigStore

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

ConfigStore loads and saves the config from a flat file.

func NewConfigStore

func NewConfigStore(basePath string) *ConfigStore

func (*ConfigStore) Load

func (s *ConfigStore) Load() Config

func (*ConfigStore) Save

func (s *ConfigStore) Save(cfg Config) error

type Consumer

type Consumer struct {
	Deployment string  `json:"deployment"`
	Container  string  `json:"container"`
	Value      float64 `json:"value"`
}

Consumer is one container's reading in a firing snapshot.

type ContainerHealth

type ContainerHealth struct {
	Container  string `json:"container"`
	Deployment string `json:"deployment"`
	Status     string `json:"status"`
}

ContainerHealth is a container's current health, tagged with its deployment.

func DockerHealthSource

func DockerHealthSource() ([]ContainerHealth, error)

DockerHealthSource reads container health via `docker ps`.

type ContainerSample

type ContainerSample struct {
	Deployment  string
	Container   string
	CPUPercent  float64
	MemoryUsage uint64
	MemoryLimit uint64
	NetworkRx   uint64
	NetworkTx   uint64
}

ContainerSample is the raw per-container reading the scraper feeds in; the store expands it into the individual semconv metric series.

func DockerStatsSource

func DockerStatsSource() ([]ContainerSample, error)

DockerStatsSource reads a point-in-time snapshot via `docker stats`, tagging each container with its compose project (deployment) from container labels.

type DeploymentRestartFunc

type DeploymentRestartFunc func(dir string) error

DeploymentRestartFunc restarts a deployment given its directory.

type ExhaustedEvent

type ExhaustedEvent struct {
	Container  string    `json:"container"`
	Deployment string    `json:"deployment"`
	Attempts   int       `json:"attempts"`
	At         time.Time `json:"at"`
}

ExhaustedEvent reports that auto-restart has stopped trying on a container. Restarting it did not fix it, so it stays unhealthy until someone intervenes.

type HealthSource

type HealthSource func() ([]ContainerHealth, error)

HealthSource returns the current health of running containers. Injectable for tests.

type HealthWatcher

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

HealthWatcher restarts running-but-unhealthy containers. It only acts on running containers (so it never revives a deployment the user intentionally stopped), only on deployments FlatRun manages, and only up to a bounded number of attempts per unhealthy streak.

func NewHealthWatcher

func NewHealthWatcher(source HealthSource, restart RestartFunc, interval, cooldown time.Duration) *HealthWatcher

func (*HealthWatcher) Events

func (w *HealthWatcher) Events() []RecoveryEvent

Events returns the recovery actions taken, most recent last. Always non-nil so it serializes to a JSON array rather than null.

func (*HealthWatcher) OnExhausted

func (w *HealthWatcher) OnExhausted(fn func(ExhaustedEvent))

OnExhausted registers a callback fired once when auto-restart gives up on a container, which is the point the watcher stops acting and an operator has to.

func (*HealthWatcher) OnRecover

func (w *HealthWatcher) OnRecover(fn func(RecoveryEvent))

OnRecover registers a callback fired after each auto-restart, so the core notification service can be told a container was recovered.

func (*HealthWatcher) Run

func (w *HealthWatcher) Run(ctx context.Context)

Run checks health on each tick until ctx is cancelled.

func (*HealthWatcher) SetEnabled

func (w *HealthWatcher) SetEnabled(on bool)

SetEnabled turns auto-restart on or off. Health is still observed either way.

func (*HealthWatcher) SetManaged

func (w *HealthWatcher) SetManaged(fn func(deployment string) bool)

SetManaged restricts auto-restart to deployments for which the predicate returns true. Health is still observed for all containers; only the restart action is scoped.

func (*HealthWatcher) Snapshot

func (w *HealthWatcher) Snapshot() []ContainerHealth

Snapshot returns the last-seen health per container.

type HostCollector

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

HostCollector periodically records a host reading into the store, giving the system-wide metrics the same time-series treatment as per-container ones.

func NewHostCollector

func NewHostCollector(store *Store, source HostSource, interval time.Duration) *HostCollector

func (*HostCollector) Run

func (c *HostCollector) Run(ctx context.Context)

Run collects on each tick until ctx is cancelled. A failing read is skipped, not fatal, so a transient error does not stop host collection.

type HostSample

type HostSample struct {
	CPUPercent    float64
	MemoryUsage   uint64
	MemoryLimit   uint64
	MemoryPercent float64
	DiskPercent   float64
}

HostSample is a single host-wide reading.

func SystemHostSource

func SystemHostSource() (HostSample, error)

SystemHostSource reads host CPU, memory and disk from the system package. Its memory figure is working-set (total minus available), which is the number that tells whether the machine is actually under memory pressure.

type HostSource

type HostSource func() (HostSample, error)

HostSource returns the current host-wide reading. Injectable so the collector is testable without reading /proc.

type Incident

type Incident struct {
	ID          string            `json:"id"`
	RuleID      string            `json:"rule_id"`
	RuleName    string            `json:"rule_name"`
	Deployment  string            `json:"deployment"`
	Service     string            `json:"service,omitempty"`
	Source      string            `json:"source,omitempty"`
	Level       string            `json:"level"`
	Fingerprint string            `json:"fingerprint"`
	Sample      string            `json:"sample"`
	Context     []string          `json:"context,omitempty"`
	Count       int               `json:"count"`
	FirstSeen   time.Time         `json:"first_seen"`
	LastSeen    time.Time         `json:"last_seen"`
	Targets     []string          `json:"targets,omitempty"`
	Triage      *Triage           `json:"triage,omitempty"`
	Responses   []ResponderResult `json:"responses,omitempty"`
}

Incident is one distinct fault, seen enough times to be worth raising.

func (Incident) Key

func (i Incident) Key() string

Key identifies the fault rather than this sighting of it, so the same crash next week has the same key. A responder keys its own work on it.

func (Incident) Message

func (i Incident) Message() string

Message leads with the triage when there is one, and the line itself when there is not.

func (Incident) Title

func (i Incident) Title() string

type LatestPoint

type LatestPoint struct {
	SeriesKey
	Sample
}

LatestPoint is a series' most recent sample.

type LogEngine

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

LogEngine runs the funnel: level, pattern, burst, fingerprint cooldown, then whatever the rule asked for. Everything before the last step is local, so a container writing a line a millisecond costs a regex per line and nothing else.

func NewLogEngine

func NewLogEngine() *LogEngine

func (*LogEngine) AttachResponses

func (e *LogEngine) AttachResponses(incidentID string, results []ResponderResult)

AttachResponses records what the responders did.

func (*LogEngine) Explain

func (e *LogEngine) Explain(incident Incident) *Triage

Explain records what the model makes of an incident. It waits on a network call, so it must not be called from the path that reads log lines.

func (*LogEngine) Incidents

func (e *LogEngine) Incidents() []Incident

func (*LogEngine) Offer

func (e *LogEngine) Offer(line LogLine) []Incident

Offer runs one line through the funnel, returning what it raised. It never leaves the machine, so it is cheap enough to call for every line read.

func (*LogEngine) OnTriage

func (e *LogEngine) OnTriage(fn func(ctx context.Context, incident Incident) (*Triage, error))

OnTriage registers what explains an incident, called only for rules that asked for it.

func (*LogEngine) Rules

func (e *LogEngine) Rules() []LogRule

func (*LogEngine) SetContextLines

func (e *LogEngine) SetContextLines(n int)

func (*LogEngine) SetRules

func (e *LogEngine) SetRules(rules []LogRule)

SetRules replaces the rule set, forgetting the state of rules that no longer exist.

type LogLine

type LogLine struct {
	Deployment string
	Service    string
	Source     string
	Level      string
	Message    string
	Raw        string
	At         time.Time
}

LogLine is one parsed line, in the shape the agent's log stream already produces.

type LogRule

type LogRule struct {
	ID         string `json:"id" yaml:"id"`
	Name       string `json:"name" yaml:"name"`
	Enabled    bool   `json:"enabled" yaml:"enabled"`
	Deployment string `json:"deployment" yaml:"deployment"`
	Service    string `json:"service,omitempty" yaml:"service,omitempty"`
	Source     string `json:"source,omitempty" yaml:"source,omitempty"`
	// Defaults to error: a rule watching info is a rule watching everything.
	MinLevel string `json:"min_level,omitempty" yaml:"min_level,omitempty"`
	// Matched against the parsed message, not the raw line, so it cannot hit a timestamp
	// or a service name.
	Pattern       string `json:"pattern,omitempty" yaml:"pattern,omitempty"`
	MinCount      int    `json:"min_count,omitempty" yaml:"min_count,omitempty"`
	WindowSeconds int    `json:"window_seconds,omitempty" yaml:"window_seconds,omitempty"`
	// Repeats inside the cooldown are counted onto the open incident rather than raising
	// another.
	CooldownSeconds int `json:"cooldown_seconds,omitempty" yaml:"cooldown_seconds,omitempty"`
	// The only field that costs money to run, so it is off unless asked for.
	Triage     bool     `json:"triage,omitempty" yaml:"triage,omitempty"`
	Responders []string `json:"responders,omitempty" yaml:"responders,omitempty"`
	Targets    []string `json:"targets,omitempty" yaml:"targets,omitempty"`
}

LogRule turns lines a deployment writes into an incident. Every field except Triage exists to make the expensive part rare, so a model is only asked about what survives all of them.

func (LogRule) Validate

func (r LogRule) Validate() error

Validate reports why a rule cannot be used, if it cannot.

func (LogRule) WithDefaults

func (r LogRule) WithDefaults() LogRule

WithDefaults fills the fields a rule may leave empty.

type LogRuleStore

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

LogRuleStore keeps log rules in a flat file beside the metric rules.

func NewLogRuleStore

func NewLogRuleStore(basePath string) *LogRuleStore

func (*LogRuleStore) Load

func (s *LogRuleStore) Load() []LogRule

func (*LogRuleStore) Save

func (s *LogRuleStore) Save(rules []LogRule) error

type LogWatcher

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

LogWatcher keeps a reader open per stream the enabled rules need and feeds the engine.

func NewLogWatcher

func NewLogWatcher(engine *LogEngine, base, token string) *LogWatcher

func (*LogWatcher) OnIncident

func (w *LogWatcher) OnIncident(fn func(incident Incident, responders []string))

func (*LogWatcher) Run

func (w *LogWatcher) Run(ctx context.Context, interval time.Duration)

Run reconciles open streams with what the rules ask for. Rules change while it runs, so this is a loop rather than one-time setup.

func (*LogWatcher) SetRules

func (w *LogWatcher) SetRules(rules func() []LogRule)

type MetricSeries

type MetricSeries struct {
	Containers []string     `json:"containers"`
	Timestamps []int64      `json:"timestamps"` // unix seconds, ascending
	Values     [][]*float64 `json:"values"`     // [container][timestamp]
}

MetricSeries is one metric's aligned time series across containers: a shared timestamp axis and, per container, a value at each timestamp (nil for gaps). This is the shape a time-series chart consumes directly.

type MetricsDB

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

MetricsDB persists samples so history survives a restart and reaches past the in-memory window. The in-memory ring stays the hot path for the live view; this backs the longer ranges and anything asked for after a restart.

func OpenMetricsDB

func OpenMetricsDB(dataDir string) (*MetricsDB, error)

OpenMetricsDB opens the metrics database beside the other FlatRun state.

func (*MetricsDB) Close

func (db *MetricsDB) Close() error

func (*MetricsDB) Maintain

func (db *MetricsDB) Maintain(stop <-chan struct{}, retention time.Duration)

Maintain rolls up and prunes on a timer until ctx is done.

func (*MetricsDB) Prune

func (db *MetricsDB) Prune(now time.Time, retention time.Duration) error

Prune drops history past the retention window.

func (*MetricsDB) Range

func (db *MetricsDB) Range(key SeriesKey, since, until time.Time) ([]Sample, error)

Range returns one series between two times, reading the resolution that suits the span.

func (*MetricsDB) Rollup

func (db *MetricsDB) Rollup(now time.Time) error

Rollup folds raw samples older than the raw window into one averaged point per step, then deletes the raw rows it replaced. Without it a busy host writes every series every few seconds forever; with it, old history costs a fraction of that and still draws the same shape at the zoom levels it is read at.

Averaging is the honest summary for these series: they are gauges, so a bucket's mean is what the container was doing over that minute.

func (*MetricsDB) Series

func (db *MetricsDB) Series(since time.Time) ([]SeriesKey, error)

Series lists the series the database holds within a window.

func (*MetricsDB) WriteBatch

func (db *MetricsDB) WriteBatch(points []LatestPoint) error

WriteBatch stores a set of samples in one transaction, which is how the collector's tick arrives and keeps a per-row fsync off the sampling path.

type RecoveryEvent

type RecoveryEvent struct {
	Container  string    `json:"container"`
	Deployment string    `json:"deployment"`
	At         time.Time `json:"at"`
}

RecoveryEvent records a self-heal action for the UI/audit.

type Responder

type Responder interface {
	Name() string
	// Respond describes what it did in a sentence; both that and any error are recorded.
	Respond(ctx context.Context, incident Incident) (string, error)
}

Responder acts on an incident: notifying, filing an issue, handing it to an agent that opens a pull request. A new one is a registration rather than a change to the engine.

One that reaches an external system should key its work on Incident.Key(), so a retry cannot produce a second issue for one fault.

func NewNotifyResponder

func NewNotifyResponder(send func(title, message string, targets []string)) Responder

NewNotifyResponder delivers the incident to the operator's configured targets.

type ResponderFunc

type ResponderFunc struct {
	ResponderName string
	Fn            func(ctx context.Context, incident Incident) (string, error)
}

func (ResponderFunc) Name

func (r ResponderFunc) Name() string

func (ResponderFunc) Respond

func (r ResponderFunc) Respond(ctx context.Context, incident Incident) (string, error)

type ResponderResult

type ResponderResult struct {
	Responder string    `json:"responder"`
	Detail    string    `json:"detail,omitempty"`
	Error     string    `json:"error,omitempty"`
	At        time.Time `json:"at"`
}

type RestartFunc

type RestartFunc func(container string) error

RestartFunc restarts a container by name. Injectable for tests.

type Sample

type Sample struct {
	Time  time.Time `json:"time"`
	Value float64   `json:"value"`
}

Sample is a single metric reading at a point in time.

type SeriesKey

type SeriesKey struct {
	Deployment string
	Container  string
	Metric     string
}

SeriesKey identifies one metric stream: a metric for a container within a deployment.

type StatsSource

type StatsSource func() ([]ContainerSample, error)

StatsSource returns the current per-container readings. It is injectable so the collector can be tested without Docker.

type Store

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

Store holds a bounded ring of recent samples per series, so the UI can render recent history without an external time-series database. Older samples are discarded once the per-series capacity is reached.

func NewStore

func NewStore(capacityPerSeries int) *Store

func (*Store) Latest

func (s *Store) Latest() []LatestPoint

Latest returns the most recent sample of every series, sorted by key.

func (*Store) OnRecord

func (s *Store) OnRecord(fn func([]LatestPoint))

OnRecord registers a sink handed exactly the samples each record wrote, so they can be persisted without re-reading the store and duplicating series that did not change.

func (*Store) Range

func (s *Store) Range(key SeriesKey, since time.Time) []Sample

Range returns the samples for a series recorded at or after since, oldest first.

func (*Store) Record

func (s *Store) Record(c ContainerSample, t time.Time)

Record expands a container reading into its semconv series and stores each at t. Network counters become a per-second rate so a threshold sees throughput, not a total that only climbs.

func (*Store) RecordHost

func (s *Store) RecordHost(h HostSample, t time.Time)

RecordHost stores a host reading under the reserved host series key.

func (*Store) Series

func (s *Store) Series() []SeriesKey

Series lists the keys currently held, so callers can enumerate what is available.

type TimeSeriesResponse

type TimeSeriesResponse struct {
	Deployment string                  `json:"deployment"`
	Metrics    map[string]MetricSeries `json:"metrics"`
}

TimeSeriesResponse is the batch of a deployment's metric series over a window.

type Triage

type Triage struct {
	Summary    string    `json:"summary,omitempty"`
	Cause      string    `json:"cause,omitempty"`
	NextStep   string    `json:"next_step,omitempty"`
	Severity   string    `json:"severity,omitempty"`
	Confidence string    `json:"confidence,omitempty"`
	Skipped    string    `json:"skipped,omitempty"`
	At         time.Time `json:"at,omitempty"`
}

Triage is what the assistant concluded. Every field is optional: an incident that was never triaged is still a complete incident.

type TriageClient

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

TriageClient asks the agent to explain an incident. The model, its key and the daily ceiling live in the agent; this decides what to send and remembers the answer.

func NewTriageClient

func NewTriageClient(base, token string) *TriageClient

func (*TriageClient) Explain

func (t *TriageClient) Explain(ctx context.Context, incident Incident) (*Triage, error)

Jump to

Keyboard shortcuts

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