observ

package
v0.4.0-beta.1 Latest Latest
Warning

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

 Go to latest Published: Jul 29, 2026 License: MIT Imports: 27 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 (
	HealthHealthy   = "healthy"
	HealthUnhealthy = "unhealthy"
	HealthStarting  = "starting"
	HealthNone      = "none"
)

Health status values as reported by Docker's HEALTHCHECK.

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.

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."},
}

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 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 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 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) 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"`
}

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"`
}

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"`
}

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 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 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 LatestPoint 

type LatestPoint struct {
	SeriesKey
	Sample
}

LatestPoint is a series' most recent sample.

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 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.

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.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.