telemetry

package
v0.8.3 Latest Latest
Warning

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

 Go to latest Published: Oct 24, 2026 License: Apache-2.0 Imports: 33 Imported by: 0

Documentation

Overview

Package telemetry is the node-local metrics store, decided in ADR 009: a dedicated SQLite database (telemetry.db, separate from internal/store's levelrail.db) holding container resource-usage samples, queried in place rather than shipped anywhere centrally. Keeping metrics node-local instead of centralizing them is what keeps idle cost near zero as the number of managed apps grows.

Index

Constants

View Source 
const (
	AudienceReferrer = "referrer"
	AudienceCountry  = "country"

	// EnvRequestAudienceMax caps distinct referrer and country labels per
	// host per tick; the rest collapse into OverflowAudience.
	EnvRequestAudienceMax = "APP_REQUEST_AUDIENCE_MAX"

	// ReferrerDirect labels requests that carry no usable Referer header.
	ReferrerDirect = "direct"
	// OverflowAudience collects labels beyond the per-tick cap.
	OverflowAudience = "other"
)

Audience kinds stored in request_audience.

View Source 
const (
	MetricAutoscaleCurrent    = "autoscale_replicas_current"
	MetricAutoscaleDesired    = "autoscale_replicas_desired"
	MetricAutoscaleNoCapacity = "autoscale_no_capacity"
	MetricAutoscaleFlaps1h    = "autoscale_flaps_1h"
)

Autoscaler gauge names, one sample per autoscaled app per controller tick.

View Source 
const (
	// EnvRequestRoutes set to "off" disables per-route request counting.
	EnvRequestRoutes = "APP_REQUEST_ROUTES"
	// EnvRequestRouteMax caps distinct routes per host per sampling tick.
	EnvRequestRouteMax = "APP_REQUEST_ROUTE_MAX"
	// EnvRequestRouteDepth caps path segments kept in a route template.
	EnvRequestRouteDepth = "APP_REQUEST_ROUTE_DEPTH"
)

Breakdown env vars.

View Source 
const (
	// MetricDeployCount is one sample per completed deploy cutover
	// (RecordDeploy), always Value 1. A raw sample count therefore *is*
	// deploy frequency for a range: Aggregate's own Count field (how
	// many raw samples fell in a bucket) already answers "how many
	// deploys in this window" with no separate counter or rate
	// computation needed.
	MetricDeployCount = "deploy_count"
	// MetricBuildDuration is one sample per completed build
	// (RecordBuildDuration), Value the build's wall-clock duration in
	// seconds, taken directly from build.Result.Duration rather than
	// re-measured here.
	MetricBuildDuration = "build_duration_seconds"
	// MetricContainerRestartCount is one sample per real container
	// restart (RecordContainerRestart), always Value 1, the same
	// "raw sample count is the metric" shape MetricDeployCount already
	// establishes above. Deliberately not sourced from Docker's own
	// container-inspect RestartCount field: every levelrail-managed
	// container is created with its restart policy disabled
	// (internal/docker's buildHostConfig), the reconciler alone decides
	// whether a dead container comes back, so Docker's own counter never
	// moves and would only ever read 0. The real signal already exists
	// in this codebase: internal/alerting.RestartTracker watches
	// Docker's event stream and counts each "start" after a container's
	// first as a restart, exactly what crashloop detection already
	// treats as ground truth; this just persists that same signal as a
	// queryable metric instead of keeping it in-memory only.
	MetricContainerRestartCount = "container_restart_count"
)

Metric names for the remaining gap: deploy frequency and build duration. Unlike every other metric in this store, neither comes from a Docker stats poll (collector.go's sampleValues): a deploy or a build is a discrete event, not a continuously observable resource, so each is recorded once, at the moment it happens, by its own caller (internal/reconcile/application for a deploy cutover, internal/deploy for a completed build) rather than by Collector.

View Source 
const (
	EnvSpikeMinRate     = "APP_TRAFFIC_SPIKE_MIN_RATE"
	EnvSpikeFactor      = "APP_TRAFFIC_SPIKE_FACTOR"
	EnvSpikeMinRequests = "APP_TRAFFIC_SPIKE_MIN_REQUESTS"
	EnvSpikeAttribution = "APP_TRAFFIC_SPIKE_ATTRIBUTION"
	EnvSpikeMax         = "APP_TRAFFIC_SPIKE_MAX"
	EnvSpikeBuckets     = "APP_TRAFFIC_SPIKE_BUCKETS"
)

Env vars that tune error spike detection; unset or invalid values fall back to defaults.

View Source 
const (
	ErrorStatusOpen     = "open"
	ErrorStatusResolved = "resolved"
	ErrorStatusIgnored  = "ignored"
)

Error issue statuses.

View Source 
const (
	MetricDiskUsedBytes  = "disk_used_bytes"
	MetricDiskTotalBytes = "disk_total_bytes"
)

MetricDiskUsedBytes and MetricDiskTotalBytes are the sample metric names HostDiskCollector writes, and the names internal/api and internal/alerting read back.

View Source 
const (
	MetricMemoryAvailableBytes = "memory_available_bytes"
	MetricMemoryTotalBytes     = "memory_total_bytes"
	// MetricSwapUsedBytes is SwapTotal minus SwapFree, written only where
	// /proc/meminfo exists.
	MetricSwapUsedBytes = "swap_used_bytes"
)

MetricMemoryAvailableBytes and MetricMemoryTotalBytes are the sample metric names HostMemoryCollector writes, and the names internal/api reads back.

View Source 
const (
	MetricOSPatchesAvailable         = "os_patches_available"
	MetricOSSecurityPatchesAvailable = "os_security_patches_available"
)

MetricOSPatchesAvailable and MetricOSSecurityPatchesAvailable are the sample metric names HostPatchCollector writes, and the names internal/api reads back (node_patch_status.go, node_metrics.go).

View Source 
const (
	MetricLBUpstreamsTotal   = "lb_upstreams_total"
	MetricLBUpstreamsHealthy = "lb_upstreams_healthy"
	MetricLBActiveRequests   = "lb_active_requests"
)

Load balancer metric names, one sample per service per sampling tick.

View Source 
const (
	OpEqual    = "="
	OpNotEqual = "!="
	OpContains = "~"
	OpGT       = ">"
	OpGTE      = ">="
	OpLT       = "<"
	OpLTE      = "<="
)

Field filter operators.

View Source 
const (
	LevelTrace = "trace"
	LevelDebug = "debug"
	LevelInfo  = "info"
	LevelWarn  = "warn"
	LevelError = "error"
	LevelFatal = "fatal"
)

Log levels in ascending severity; an empty level means the line carries none.

View Source 
const (
	PipelineMultiline = "multiline"
	PipelineRedact    = "redact"
	PipelineParse     = "parse"
	PipelineDrop      = "drop"
)

Pipeline rule kinds.

View Source 
const (
	LogQLLabelApp    = "app"
	LogQLLabelStream = "stream"
	LogQLLabelLevel  = "level"
)

LogQL label names the node-local store can answer for.

View Source 
const (
	MetricHTTPRequests       = "http_requests"
	MetricHTTPResponses2xx   = "http_responses_2xx"
	MetricHTTPResponses3xx   = "http_responses_3xx"
	MetricHTTPResponses4xx   = "http_responses_4xx"
	MetricHTTPResponses5xx   = "http_responses_5xx"
	MetricHTTPUpstreamErrors = "http_upstream_errors"
	MetricHTTPBytesIn        = "http_bytes_in"
	MetricHTTPBytesOut       = "http_bytes_out"

	// AbuseReasonCount mirrors abuse.ReasonCount; a test in internal/ingress
	// keeps the two in step without an import.
	AbuseReasonCount = 4
	// ShieldReasonCount mirrors trafficpolicy.ShieldReasonCount; a test in
	// internal/ingress keeps the two in step without an import.
	ShieldReasonCount = 4
)

Request (RED) metric names. All are per-tick counter deltas stored under the app's "service:<name>" resource id; only non-zero values are written, so an app with no traffic writes nothing. Every name with the "http_" prefix is a sum-kind metric: rollups add it up instead of averaging.

View Source 
const (
	MetricAbuseAllowed  = metricHTTPAbusePrefix + "allowed"
	MetricAbuseFailOpen = metricHTTPAbusePrefix + "fail_open"
)

Abuse metric names. The per-reason ones are suffixed with AbuseReasons.

View Source 
const (
	// EnvRequestUpstreamMax caps distinct upstream labels per host per tick.
	EnvRequestUpstreamMax = "APP_REQUEST_UPSTREAM_MAX"
	// EnvRequestSampleRate is the fraction (0..1) of slow requests kept in the
	// sample log; failed requests are always eligible. Default 0.1.
	EnvRequestSampleRate = "APP_REQUEST_SAMPLE_RATE"
	// EnvRequestSlowMS is the duration at which a request counts as slow.
	EnvRequestSlowMS = "APP_REQUEST_SLOW_MS"
	// EnvRequestSampleTickMax caps samples kept per app per sampling tick.
	EnvRequestSampleTickMax = "APP_REQUEST_SAMPLE_TICK_MAX"
	// EnvRequestSampleRetention is how long samples are kept (Go duration).
	EnvRequestSampleRetention = "APP_REQUEST_SAMPLE_RETENTION"
	// EnvRequestSampleRows caps stored samples per app.
	EnvRequestSampleRows = "APP_REQUEST_SAMPLE_ROWS"

	// OverflowUpstream collects traffic beyond the per-tick upstream cap.
	OverflowUpstream = "other"
)

Traffic detail env vars.

View Source 
const DefaultErrorEventsPerIssue = 20

DefaultErrorEventsPerIssue is how many sample events one issue keeps.

View Source 
const DefaultMaxPoints = 600

DefaultMaxPoints is the series cap when EnvMaxPoints is unset.

View Source 
const (

	// DefaultRedactReplacement replaces a redact match with no Replace set.
	DefaultRedactReplacement = "[REDACTED]"
)
View Source 
const EnvMaxPoints = "APP_METRICS_MAX_POINTS"

EnvMaxPoints names the env var that caps points per series.

View Source 
const EnvTrafficRetention = "APP_TRAFFIC_RETENTION"

EnvTrafficRetention overrides how long per-app traffic aggregates are kept (Go duration); unset means the minute tier retention.

View Source 
const LatencyBucketCount = 12

LatencyBucketCount is the number of histogram slots including overflow.

View Source 
const MaxPointsHardCap = 5000

MaxPointsHardCap bounds what a caller may ask for.

View Source 
const MinRequestStep = 15 * time.Second

MinRequestStep is the finest step a request series is bucketed at.

View Source 
const (

	// OverflowRoute collects traffic beyond the per-tick route cap.
	OverflowRoute = "other"
)

Variables

View Source 
var AbuseReasons = [AbuseReasonCount]string{"rate_limit", "deny_cidr", "deny_user_agent", "missing_user_agent"}

AbuseReasons are the metric suffixes, indexed like abuse.Reason.

View Source 
var ErrHostMemoryUnsupported = errors.New("telemetry: host memory is not reported on this operating system")

ErrHostMemoryUnsupported means this operating system has no memory source the control plane knows how to read.

View Source 
var ErrInvalidLogSearch = errors.New("telemetry: invalid log search")

ErrInvalidLogSearch marks a log search expression that cannot be run.

View Source 
var ErrNoErrorIssue = errors.New("telemetry: error issue not found")

ErrNoErrorIssue is returned when an issue does not exist for the app.

View Source 
var ErrNoSupportedPackageManager = errors.New("telemetry: no supported package manager found")

ErrNoSupportedPackageManager means none of apt, dnf, or yum exist on this host: a real, permanent condition (e.g. a non-Linux dev machine, or a distribution this checker doesn't know), not a transient failure.

View Source 
var LatencyBucketsMS = []float64{5, 10, 25, 50, 100, 250, 500, 1000, 2500, 5000, 10000}

LatencyBucketsMS are the fixed upper bounds (milliseconds) of the request latency histogram. A final implicit bucket holds everything above the last.

View Source 
var ShieldReasons = [ShieldReasonCount]string{"rate_limit", "concurrency", "bad_bot", "body_too_large"}

ShieldReasons are the metric suffixes, indexed like trafficpolicy.ShieldReason.

Functions

func AbuseBlockedMetric 

func AbuseBlockedMetric(i int) string

AbuseBlockedMetric counts requests enforce turned away for reason i.

func AbuseWouldBlockMetric 

func AbuseWouldBlockMetric(i int) string

AbuseWouldBlockMetric counts requests log_only would have turned away for reason i.

func AudienceMaxFromEnv 

func AudienceMaxFromEnv() int

AudienceMaxFromEnv returns the per-tick audience label cap.

func AutoStep added in v0.2.0 

func AutoStep(span time.Duration, maxPoints int, floor time.Duration) time.Duration

AutoStep picks the smallest round step, no finer than floor, that keeps a series over span within maxPoints. A non-positive maxPoints or span yields floor.

func ClassifyLevel added in v0.2.0 

func ClassifyLevel(e LogEntry) string

ClassifyLevel returns the entry's level: from the structured level/severity field when the line is JSON, else from a level token near the start of a plain line. Empty means unknown.

func DefaultLogBudgetBytes 

func DefaultLogBudgetBytes() int64

DefaultLogBudgetBytes reads APP_LOG_BUDGET_MB; 0 or unset means unlimited.

func ForecastDays 

func ForecastDays(used, budget int64, ratePerMin float64) *float64

ForecastDays estimates days until used reaches budget at ratePerMin; nil when unlimited or idle, zero when already at or over budget.

func HostDiskBytes 

func HostDiskBytes(path string) (total, used int64, err error)

HostDiskBytes reads path's filesystem total and used bytes.

func HostInodes 

func HostInodes(path string) (total, free uint64, err error)

HostInodes reads path's filesystem inode total and free counts. Filesystems without inode accounting report zero totals.

func HostMemoryBytes added in v0.2.0 

func HostMemoryBytes() (totalBytes, availableBytes int64, err error)

HostMemoryBytes reads this host's total and available memory, the same read HostMemoryCollector.CollectOnce performs on a poll tick: from /proc/meminfo where it exists, from sysctl on macOS. Exported for a one-shot caller (internal/api's ram doctor check) that has no running collector to read back from.

func HostSwapBytes 

func HostSwapBytes() (total, free int64, err error)

HostSwapBytes reads swap total and free bytes from /proc/meminfo. A host with no /proc/meminfo (macOS) or no swap reports zeros and no error.

func IsSumMetric added in v0.2.0 

func IsSumMetric(metric string) bool

IsSumMetric reports whether rollups must add the metric's values.

func LatencyBucketIndex 

func LatencyBucketIndex(ms float64) int

LatencyBucketIndex is the histogram slot for a latency in milliseconds.

func LatencyBucketMetric added in v0.2.0 

func LatencyBucketMetric(i int) string

LatencyBucketMetric is the metric name for histogram slot i.

func LevelRank added in v0.2.0 

func LevelRank(level string) int

LevelRank orders a normalised level; unknown or empty is 0.

func MarkNewSignatures added in v0.2.0 

func MarkNewSignatures(current, baseline []Signature) (newCount int)

MarkNewSignatures sets New on every signature in current that does not appear in baseline.

func MatchFieldFilters added in v0.2.0 

func MatchFieldFilters(e LogEntry, filters []FieldFilter) bool

MatchFieldFilters reports whether a line satisfies every filter. A line that is not structured, or lacks the key, matches only the != operator.

func MaxPointsFromEnv added in v0.2.0 

func MaxPointsFromEnv() int

MaxPointsFromEnv returns the configured per-series point cap.

func Normalize added in v0.2.0 

func Normalize(msg string) string

Normalize reduces a log message to its template: timestamps, ids, numbers, addresses and hashes collapse to placeholders so repeated occurrences of one error share one string.

func NormalizeLevel added in v0.2.0 

func NormalizeLevel(s string) string

NormalizeLevel maps a level word to one of the Level constants, or "".

func NormalizeRoute added in v0.2.0 

func NormalizeRoute(path string, depth int) string

NormalizeRoute reduces a URL path to a bounded template: query strings are never passed in, at most depth segments are kept, and any segment that looks like an identifier (digits, hex, UUID, long or mixed alphanumerics) becomes ":id" so tokens and user ids never reach storage.

func NumericLevel added in v0.2.0 

func NumericLevel(n float64) string

NumericLevel maps a pino/bunyan style numeric level to a Level constant.

func PercentileFromBuckets added in v0.2.0 

func PercentileFromBuckets(counts []float64, q float64) float64

PercentileFromBuckets estimates the q-quantile (0..1) in milliseconds from histogram counts, interpolating linearly inside the containing bucket. The overflow bucket reports the last finite bound. Empty input returns 0.

func QuoteLogPhrase 

func QuoteLogPhrase(text string) string

QuoteLogPhrase turns text into a search expression that matches it as one literal phrase.

func RedactLine added in v0.2.0 

func RedactLine(s string) string

RedactLine masks credentials in a log line before it leaves the store: key=value secrets, bearer tokens, URL passwords and well known key shapes.

func ReferrerHost 

func ReferrerHost(referer string) string

ReferrerHost reduces a Referer header to a lowercase host, or "" when it is empty or unparseable. The path and query are dropped, never stored.

func RequestOverviews added in v0.2.0 

func RequestOverviews(ctx context.Context, q MetricsQuerier, apps []string, window, step time.Duration, now time.Time) (map[string]RequestOverview, error)

RequestOverviews computes SummarizeRequests and QueryRequests for every app over [now-window, now] from one batched sample read when q supports it, falling back to the per app functions otherwise.

func RouteDepthFromEnv added in v0.2.0 

func RouteDepthFromEnv() int

RouteDepthFromEnv returns how many path segments a route template keeps.

func RouteMaxFromEnv added in v0.2.0 

func RouteMaxFromEnv() int

RouteMaxFromEnv returns the per-tick route cap.

func RoutesEnabled added in v0.2.0 

func RoutesEnabled() bool

RoutesEnabled reports whether per-route counting is on.

func SampleRetentionFromEnv 

func SampleRetentionFromEnv() time.Duration

SampleRetentionFromEnv returns how long request samples are kept.

func SelfRSSBytes added in v0.2.0 

func SelfRSSBytes() uint64

SelfRSSBytes is this process's resident set size. Falls back to the Go runtime's footprint where /proc is unavailable.

func ShieldMetric 

func ShieldMetric(i int) string

ShieldMetric is the metric name counting requests turned away for reason i.

func SignatureOf added in v0.2.0 

func SignatureOf(e LogEntry) (id, template string)

SignatureOf returns the stable id and template of an entry's message.

func TrafficRetentionFromEnv 

func TrafficRetentionFromEnv(def time.Duration) time.Duration

TrafficRetentionFromEnv returns the traffic aggregate retention, or def.

func UpstreamMaxFromEnv 

func UpstreamMaxFromEnv() int

UpstreamMaxFromEnv returns the per-tick upstream label cap.

func ValidateLogSearch 

func ValidateLogSearch(query string) error

ValidateLogSearch reports whether query is a runnable search expression.

func ValidatePipeline 

func ValidatePipeline(rules []PipelineRule) error

ValidatePipeline checks kinds and that every pattern compiles.

Types

type AbuseWindow 

type AbuseWindow struct {
	Allowed    uint64
	WouldBlock [AbuseReasonCount]uint64
	Blocked    [AbuseReasonCount]uint64
	FailOpen   uint64
}

AbuseWindow counts one tick of the per-app ingress abuse defense. WouldBlock is what log_only saw; Blocked is what enforce turned away. FailOpen counts requests served unchecked because the defense itself failed.

type AggregatedPoint 

type AggregatedPoint struct {
	// Timestamp is the bucket's start, not a sample's own timestamp.
	Timestamp time.Time
	// Value is the average of every sample that fell in this bucket.
	Value float64
	// Count is how many raw samples contributed, so a caller (or a
	// future frontend) can tell a bucket built from one sample from one
	// built from sixty, without a separate query.
	Count int
	// Max is the largest sample in the bucket, so a spike survives
	// downsampling even though Value is an average.
	Max float64
}

AggregatedPoint is one bucketed value from Aggregate.

func Aggregate 

func Aggregate(samples []Sample, from time.Time, step time.Duration) []AggregatedPoint

Aggregate buckets samples (assumed already timestamp-ascending, which Federator.QueryMetrics already guarantees) into fixed-width windows starting at from, each bucket's value the average of the samples inside it. step <= 0 means "no aggregation," one point per sample. An empty bucket is omitted, not returned as a zero: a sparse series (a container that stopped reporting for an hour) should read as a gap in the chart, not a real reading of zero.

func ShiftPoints added in v0.2.0 

func ShiftPoints(pts []AggregatedPoint, d time.Duration) []AggregatedPoint

ShiftPoints returns pts with every timestamp moved forward by d, so a previous-period series lines up with the current window.

type AudienceRow 

type AudienceRow struct {
	Kind     string
	Label    string
	Requests float64
}

AudienceRow is one referrer host or country and its request count.

type AudienceSource 

type AudienceSource interface {
	QueryAudience(ctx context.Context, resourceID string, from, to time.Time) ([]AudienceRow, error)
}

AudienceSource reads referrer and country totals; *DB satisfies it.

type AutoscaleSample 

type AutoscaleSample struct {
	App        string
	Current    int
	Desired    int
	NoCapacity bool
	Flaps      int
}

AutoscaleSample is one app's autoscaler snapshot.

type BoundedLogQuerier added in v0.2.0 

type BoundedLogQuerier interface {
	QueryLogs(ctx context.Context, resourceID string, from, to time.Time, query string) ([]LogEntry, error)
}

BoundedLogQuerier is the unbounded read every log source has.

type Breakdown added in v0.2.0 

type Breakdown struct {
	Routes   []RouteRow
	Statuses []StatusRow
}

Breakdown is the route and status totals for one app and window.

func (Breakdown) SortedStatuses added in v0.2.0 

func (b Breakdown) SortedStatuses() []StatusRow

SortedStatuses returns status totals by count descending, ties by code.

func (Breakdown) TopRoutes added in v0.2.0 

func (b Breakdown) TopRoutes(n int) []RouteRow

TopRoutes returns the n busiest routes, requests descending, ties by name.

func (Breakdown) Total added in v0.2.0 

func (b Breakdown) Total() float64

Total returns the request count, preferring exact status totals.

type BreakdownSource added in v0.2.0 

type BreakdownSource interface {
	QueryBreakdown(ctx context.Context, resourceID string, from, to time.Time) (Breakdown, error)
}

BreakdownSource reads route and status totals; *DB satisfies it.

type Collector 

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

Collector polls a caller-supplied set of targets on an interval and writes what it collects to a Store.

func NewCollector 

func NewCollector(source StatsSource, store *DB, interval time.Duration, logger *slog.Logger) *Collector

NewCollector builds a Collector. logger defaults to slog.Default() if nil.

func (*Collector) CollectOnce 

func (c *Collector) CollectOnce(ctx context.Context, targets []Target) error

CollectOnce polls every target once and writes whatever succeeded. One target's stats call failing (its container just stopped between discovery and polling, for instance) doesn't block the others': the same "one broken resource must not stop convergence of everything else" principle reconcile.Engine.ReconcileAll already applies to controllers, applied here to collection targets. Returns a joined error of every target that failed, purely for the caller to log or count; a partial failure is not treated as this tick failing overall, since the successful samples were still written.

func (*Collector) Run 

func (c *Collector) Run(ctx context.Context, targetsFunc func(context.Context) ([]Target, error)) error

Run calls CollectOnce every interval until ctx is done. targetsFunc is invoked fresh on every tick, not captured once at construction, so the polled set stays current as services are created, redeployed, or removed between ticks: the same level-triggered, re-derive-every-pass principle every reconcile.Controller already follows, applied here to "what should be collected" instead of "what should be running."

type DB 

type DB struct {
	*sql.DB
	// contains filtered or unexported fields
}

DB wraps a *sql.DB opened against telemetry.db: WAL mode, migrated to the latest version before Open returns. Deliberately a separate type from store.DB, not a shared one: ADR 009 keeps this on its own SQLite file precisely so its write pattern (a batch every collection tick) never contends with internal/store's own WAL.

func Open 

func Open(ctx context.Context, path string) (*DB, error)

Open opens (creating if needed) the SQLite database at path, applies pragmas, and runs every pending migration. path is a plain filesystem path, not a DSN, matching internal/store.Open's own shape.

func (*DB) Compact 

func (db *DB) Compact(ctx context.Context) error

Compact prunes past retention and the size cap, then returns freed pages to the filesystem when more than a quarter of the file is free.

func (*DB) Configure added in v0.2.0 

func (db *DB) Configure(cfg TierConfig)

Configure sets the tier configuration. Call it before the DB is shared with other goroutines.

func (*DB) DeleteLogBudget 

func (db *DB) DeleteLogBudget(ctx context.Context, resourceID string) error

DeleteLogBudget removes the override so the global default applies again.

func (*DB) DeleteLogPipeline 

func (db *DB) DeleteLogPipeline(ctx context.Context, resourceID string) error

DeleteLogPipeline removes the rules for resourceID.

func (*DB) DistinctLogResources added in v0.2.0 

func (db *DB) DistinctLogResources(ctx context.Context, fromNs, toNs int64) ([]string, error)

DistinctLogResources lists resources with at least one log line in [fromNs, toNs).

func (*DB) EnforceLogBudgets 

func (db *DB) EnforceLogBudgets(ctx context.Context, defaultMax int64) ([]LogBudgetTrim, error)

EnforceLogBudgets trims the oldest rows of every resource over its budget (override, else defaultMax). Only that resource's rows are touched, and each resource drains at most trimMaxBatches batches per call.

func (*DB) FileBytes 

func (db *DB) FileBytes(ctx context.Context) (int64, error)

FileBytes returns the size of the database file including free pages.

func (*DB) GetErrorIssue 

func (db *DB) GetErrorIssue(ctx context.Context, resourceID, fingerprint string) (ErrorIssue, []ErrorSample, error)

GetErrorIssue returns one issue and its stored samples, newest first.

func (*DB) GetLogBudget 

func (db *DB) GetLogBudget(ctx context.Context, resourceID string) (int64, error)

GetLogBudget returns the stored override in bytes, 0 when none.

func (*DB) GetLogPipeline 

func (db *DB) GetLogPipeline(ctx context.Context, resourceID string) ([]PipelineRule, error)

GetLogPipeline returns the rules for resourceID, nil when none are set.

func (*DB) GetTrace 

func (db *DB) GetTrace(ctx context.Context, resourceID, traceID string) ([]Span, error)

GetTrace returns every stored span of a trace, earliest first, optionally limited to one resource.

func (*DB) LatestByMetric added in v0.2.0 

func (db *DB) LatestByMetric(ctx context.Context, metric string) ([]Sample, error)

LatestByMetric returns the most recent sample for every resource_id that has ever recorded metric, one row per resource: the "what is everything doing right now" read a dashboard-wide ranking needs (e.g. "which app is using the most CPU"), where fetching each resource's own time series one at a time would mean one query per app instead of one query total. An empty (nil) result and a nil error mean "no resource has recorded this metric yet," the same convention Query's own doc comment establishes.

func (*DB) LatestLogTimestamp added in v0.2.0 

func (db *DB) LatestLogTimestamp(ctx context.Context, resourceID, containerID string) (time.Time, error)

LatestLogTimestamp returns the newest stored line time for containerID, or the zero time when none is stored.

func (*DB) ListErrorIssues 

func (db *DB) ListErrorIssues(ctx context.Context, f ErrorFilter) ([]ErrorIssue, error)

ListErrorIssues returns the app's issues, most recently seen first.

func (*DB) ListTraces 

func (db *DB) ListTraces(ctx context.Context, f TraceFilter) ([]TraceSummary, error)

ListTraces returns the newest matching traces first. A trace is matched by any of its spans being in the window, and its duration spans all of its stored spans.

func (*DB) LogUsageAll 

func (db *DB) LogUsageAll(ctx context.Context, now time.Time, defaultMax int64) ([]LogUsage, error)

LogUsageAll returns usage for every resource that has log rows, an override, or trim history, noisiest (most stored bytes) first. It runs two grouped aggregates (all rows, and the last 24h) plus two tiny table reads.

func (*DB) Prune added in v0.2.0 

func (db *DB) Prune(ctx context.Context, now time.Time) (PruneResult, error)

Prune deletes samples and rollups past their retention and then, while the database exceeds MaxBytes, the oldest tenth of the finest tier that still has data already covered by the next tier. Raw rows are never dropped for size before they are rolled up.

func (*DB) Query 

func (db *DB) Query(ctx context.Context, resourceID, metric string, from, to time.Time) ([]Sample, error)

Query returns samples for resourceID/metric in [from, to], oldest first, reading the tier the range calls for. Rollup tiers return one point per bucket (sum for counters, average for gauges) and are topped up with the finer tier for the not yet rolled up tail.

func (*DB) QueryAudience 

func (db *DB) QueryAudience(ctx context.Context, resourceID string, from, to time.Time) ([]AudienceRow, error)

QueryAudience sums referrer and country counts for resourceID in [from, to].

func (*DB) QueryBreakdown added in v0.2.0 

func (db *DB) QueryBreakdown(ctx context.Context, resourceID string, from, to time.Time) (Breakdown, error)

QueryBreakdown sums route and status counters for resourceID in [from, to].

func (*DB) QueryDeployLog 

func (db *DB) QueryDeployLog(ctx context.Context, attemptID string) ([]DeployLogEntry, error)

QueryDeployLog returns every persisted line for attemptID, oldest first: a full replay, not a windowed range query like QueryLogs, since a deploy attempt's log is a bounded, already-finished (by the time this is the code path serving it, see internal/api/deploys.go's SSE handler) event with no reason to page through it by time range. An empty (nil) result and a nil error both mean "no lines for this attempt," the same "absence is not an error" convention QueryLogs already establishes, covering both a genuinely empty log (the plain image-tag trigger path, which never calls WriteDeployLogBatch at all) and an attempt ID that was never written for any reason.

func (*DB) QueryLogs 

func (db *DB) QueryLogs(ctx context.Context, resourceID string, from, to time.Time, query string) ([]LogEntry, error)

QueryLogs returns every log entry for resourceID in [from, to], oldest first, optionally filtered by a search expression (see compileLogSearch: phrases, prefix*, AND, OR, NOT, -word, parentheses). An empty query returns every line in range. A malformed expression returns ErrInvalidLogSearch.

An empty (nil) result and a nil error both mean "no lines matched," not an error, the same convention Query already establishes for metric samples.

func (*DB) QueryLogsLimited added in v0.2.0 

func (db *DB) QueryLogsLimited(ctx context.Context, resourceID string, from, to time.Time, query string, limit int) ([]LogEntry, bool, error)

QueryLogsLimited returns the newest limit lines for resourceID in [from, to], oldest first, and whether older matching lines were left out. It reads through idx_log_entries_resource_ts and stops at limit+1 rows, so a rule or aggregate over a busy app never scans the whole window.

func (*DB) QueryMany added in v0.2.0 

func (db *DB) QueryMany(ctx context.Context, resourceIDs, metrics []string, from, to time.Time) ([]Sample, error)

QueryMany returns every sample for the resource and metric cross product in [from, to] as one statement. Ranges the raw tier does not serve fall back to per series Query so tier selection stays in one place.

func (*DB) QueryRequestSamples 

func (db *DB) QueryRequestSamples(ctx context.Context, resourceID string, from, to time.Time, limit int) ([]RequestSample, error)

QueryRequestSamples returns up to limit samples in [from, to], newest first.

func (*DB) QueryUpstreams 

func (db *DB) QueryUpstreams(ctx context.Context, resourceID string, from, to time.Time) ([]UpstreamRow, error)

QueryUpstreams sums per-upstream counters for resourceID in [from, to].

func (*DB) RecordAutoscale 

func (db *DB) RecordAutoscale(ctx context.Context, samples []AutoscaleSample, at time.Time) error

RecordAutoscale writes the autoscaler gauges for each sample, so a plain threshold alert rule can watch NoCapacity and flapping.

func (*DB) RecordBuildDuration 

func (db *DB) RecordBuildDuration(ctx context.Context, serviceName string, d time.Duration, at time.Time) error

RecordBuildDuration writes one MetricBuildDuration sample for serviceName at at.

func (*DB) RecordContainerRestart added in v0.2.0 

func (db *DB) RecordContainerRestart(ctx context.Context, serviceName string, at time.Time) error

RecordContainerRestart writes one MetricContainerRestartCount sample for serviceName at at. Callers must call this only for a genuine restart of an already-started container (alerting.RestartTracker's own "not the container's first start" definition), not for a fresh deploy's first start.

func (*DB) RecordDeploy 

func (db *DB) RecordDeploy(ctx context.Context, serviceName string, at time.Time) error

RecordDeploy writes one MetricDeployCount sample for serviceName at at. Callers must call this only when a deploy actually happened (a new container was created/started and passed its readiness probe), not on every reconcile tick that finds nothing to do: internal/reconcile/ application.Controller's own "Deployed" vs "AlreadyRunning" distinction is exactly that signal.

func (*DB) RecordLoadBalancer added in v0.2.0 

func (db *DB) RecordLoadBalancer(ctx context.Context, samples []LoadBalancerSample, at time.Time) error

RecordLoadBalancer writes the three load balancer gauges for each sample.

func (*DB) RecordRequests added in v0.2.0 

func (db *DB) RecordRequests(ctx context.Context, byApp map[string]RequestWindow, at time.Time) error

RecordRequests writes the per-app windows as samples at time at.

func (*DB) Retain 

func (db *DB) Retain(ctx context.Context, cutoff time.Time) (deleted int64, err error)

Retain deletes every sample older than the given cutoff, returning how many rows were removed. The documented default retention (15 days) is a caller concern (the collector or a scheduled sweep decides the cutoff), not hardcoded here, per the house "no hardcoded thresholds" rule.

func (*DB) RetainDeployLogs 

func (db *DB) RetainDeployLogs(ctx context.Context, cutoff time.Time) (deleted int64, err error)

RetainDeployLogs deletes every deploy log line older than cutoff, returning how many rows were removed. Mirrors RetainLogs' shape exactly, same "caller decides the cutoff, no hardcoded threshold" house rule; see cmd/levelrail/main.go's retention sweep wiring for the default this project actually runs with, chosen separately from RetainLogs' 15-day container-log default per this table's own "bounded, terminal event" reasoning (migrations/0003_deploy_logs.sql).

func (*DB) RetainErrors 

func (db *DB) RetainErrors(ctx context.Context, cutoff time.Time) (int64, error)

RetainErrors deletes events and issues last seen before cutoff.

func (*DB) RetainLogs 

func (db *DB) RetainLogs(ctx context.Context, cutoff time.Time) (deleted int64, err error)

RetainLogs deletes every log entry older than cutoff, returning how many rows were removed. Mirrors Retain's shape exactly (same signature pattern, same "caller decides the cutoff, no hardcoded threshold" house rule); the FTS index stays in sync automatically via log_entries_ad (migrations/0002_log_entries.sql), not because of anything this method does itself.

func (*DB) RetainSpans 

func (db *DB) RetainSpans(ctx context.Context, cutoff time.Time) (int64, error)

RetainSpans deletes spans that started before cutoff.

func (*DB) Rollup added in v0.2.0 

func (db *DB) Rollup(ctx context.Context, now time.Time) (RollupResult, error)

Rollup computes closed 1 minute buckets from raw samples and closed 1 hour buckets from the minute tier, at most MaxBucketsPerRun buckets per tier. Each chunk and its watermark commit in one transaction, and recomputing a bucket is idempotent, so an interrupted run resumes where it stopped.

func (*DB) SetClock added in v0.2.0 

func (db *DB) SetClock(now func() time.Time)

SetClock replaces the time source used for tier selection and maintenance; tests use it to drive a fake clock.

func (*DB) SetErrorIssueStatus 

func (db *DB) SetErrorIssueStatus(ctx context.Context, resourceID, fingerprint, status string) error

SetErrorIssueStatus changes an issue's status and clears its regression flag.

func (*DB) SetLogBudget 

func (db *DB) SetLogBudget(ctx context.Context, resourceID string, maxBytes int64) error

SetLogBudget stores the override; maxBytes <= 0 clears it.

func (*DB) SetLogPipeline 

func (db *DB) SetLogPipeline(ctx context.Context, resourceID string, rules []PipelineRule) error

SetLogPipeline validates and stores the rules; an empty list clears them.

func (*DB) StreamLogs added in v0.2.0 

func (db *DB) StreamLogs(ctx context.Context, resourceID string, fromNs, toNs int64, maxLines int, fn func(LogEntry) error) (LogStreamResult, error)

StreamLogs calls fn for each line of resourceID in [fromNs, toNs) in timestamp order, at most maxLines. When the cap is hit the range ends at a timestamp boundary so a follow-up call starting at EndNs neither repeats nor skips lines.

func (*DB) TrimLogBudget 

func (db *DB) TrimLogBudget(ctx context.Context, resourceID string, defaultMax int64) (LogBudgetTrim, bool, error)

TrimLogBudget trims one resource's oldest rows down to its budget (override, else defaultMax) now. The bool reports a backlog larger than one pass drains.

func (*DB) WriteDeployLogBatch 

func (db *DB) WriteDeployLogBatch(ctx context.Context, entries []DeployLogEntry) error

WriteDeployLogBatch inserts every entry in one transaction, the same atomicity and "no upsert, a retried batch can produce duplicate rows" tradeoff WriteLogBatch already documents for the identical reason: a deploy attempt's output has no natural idempotency key either. internal/deploylog.Recorder is this method's only caller, and it already batches by count (see that package's own batchMaxLines) before calling this, matching WriteLogBatch's batching discipline.

func (*DB) WriteErrorEvent 

func (db *DB) WriteErrorEvent(ctx context.Context, ev ErrorEvent, keep int) (regressed bool, err error)

WriteErrorEvent records ev against its issue, creating the issue or bumping its count, and trims the issue's samples to keep. A resolved issue that receives an event reopens flagged as a regression, reported as regressed.

func (*DB) WriteLogBatch 

func (db *DB) WriteLogBatch(ctx context.Context, entries []LogEntry) error

WriteLogBatch inserts every entry in one transaction, the same atomicity WriteSamples already gives metric collection ticks, applied here to one log batch (see logBatchMaxLines/logBatchMaxWait). Unlike WriteSamples, there's no ON CONFLICT upsert: metric samples have a natural idempotency key (resource, metric, timestamp) a retried collection tick can safely overwrite, but log lines don't, since a container legitimately emitting the same text twice (e.g. two "OK" lines) is normal, not a duplicate to be coalesced. This is a real, documented gap: if a batch write fails partway and the caller decides to retry with the same lines, retried lines land as new rows rather than replacing anything, so a transient write failure during streaming can produce duplicate log rows in rare cases. Closing that gap would need its own idempotency key (e.g. a source-assigned sequence number per container), deliberately deferred rather than solved silently here.

func (*DB) WriteSamples 

func (db *DB) WriteSamples(ctx context.Context, samples []Sample) error

WriteSamples inserts every sample in one transaction: a collection tick writes many samples (several metrics per running container) and they should land atomically, not partially, if the process is interrupted mid-write. Writing the same (resource, metric, timestamp) twice replaces the value rather than erroring, since a collector that retries a tick after a transient failure should be safe to just run again, not required to first check what already landed.

func (*DB) WriteSpans 

func (db *DB) WriteSpans(ctx context.Context, spans []Span) error

WriteSpans stores spans, ignoring any whose (trace, span) id is already there so a retried export is harmless.

type DeployLogEntry 

type DeployLogEntry struct {
	AttemptID string
	// Stream is "stdout" or "stderr".
	Stream    string
	Timestamp time.Time
	Message   string
}

DeployLogEntry is one line of a deploy attempt's build output.

type DeployMarker 

type DeployMarker struct {
	ID     string
	At     time.Time
	Status string
	Image  string
	Commit string
}

DeployMarker is a deploy the detector may blame a spike on.

type DrainConfig 

type DrainConfig struct {
	ResourceID string
	Type       DrainSinkType
	Target     string
	Enabled    bool
}

DrainConfig is one resource's external log-forwarding configuration, the telemetry-package-local shape a caller's configFunc (see Run) resolves from wherever it actually persists (internal/store's LogDrain, for application services).

type DrainForwarder 

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

DrainForwarder taps LogBroadcaster the same way a live SSE viewer does (Subscribe/Publish, broadcast.go), so it is purely an additional consumer of the log stream: it never touches LogCollector, WriteLogBatch, or QueryLogs, and a drain forwarding failure can never affect the node-local store or a live tail.

func NewDrainForwarder 

func NewDrainForwarder(broadcaster *LogBroadcaster, httpClient *http.Client, logger *slog.Logger, programTag string) *DrainForwarder

NewDrainForwarder builds a DrainForwarder reading from broadcaster. logger defaults to slog.Default() if nil. programTag is the syslog program tag a DrainSinkSyslog sink identifies itself with (see newSyslogSink); callers pass brand.Brand.ShortName, following the same "resolve brand outside this package" convention internal/reconcile/application.WithNetworkPrefix already establishes.

func (*DrainForwarder) Run 

func (f *DrainForwarder) Run(ctx context.Context, resync time.Duration, configFunc func(context.Context) ([]DrainConfig, error)) error

Run derives the desired drain set via configFunc on a resync interval and keeps exactly one forwarding goroutine alive per currently-enabled ResourceID, mirroring LogCollector.Run's own reconcile-loop shape exactly (same active-map/start/cancel pattern) so the two collectors read as one family even though they serve different consumers.

type DrainSink 

type DrainSink interface {
	Send(ctx context.Context, entries []LogEntry) error
}

DrainSink is the narrow surface a log drain forwarder needs from a concrete sink implementation (HTTPSink, syslog). Send should treat a partial failure (e.g. some lines rejected) as a whole-batch error: there's no per-line retry here, a failed batch is simply dropped and logged, the same best-effort shape LogBroadcaster.Publish's own doc comment already establishes for a slow live-tail subscriber.

type DrainSinkType 

type DrainSinkType string

DrainSinkType selects which external sink protocol a drain uses. Defined locally rather than importing internal/store's LogDrainType: this package's own doc comments elsewhere (e.g. DesiredService's NodeStatus precedent) already establish "define locally, don't import a higher-level package's vocabulary" for cross-package enums like this. 

const (
	DrainSinkHTTP   DrainSinkType = "http"
	DrainSinkSyslog DrainSinkType = "syslog"
)

The two sink protocols buildDrainSink knows how to construct.

type ErrorEvent 

type ErrorEvent struct {
	ResourceID  string
	Fingerprint string
	Title       string
	Level       string
	EventID     string
	TimestampNs int64
	Release     string
	Environment string
	DeployID    string
	Raw         string
}

ErrorEvent is one received error, already grouped by Fingerprint.

type ErrorFilter 

type ErrorFilter struct {
	ResourceID string
	Status     string
	Since      time.Time
	Limit      int
}

ErrorFilter narrows ListErrorIssues.

type ErrorIssue 

type ErrorIssue struct {
	Fingerprint  string `json:"fingerprint"`
	Title        string `json:"title"`
	Level        string `json:"level"`
	FirstSeenNs  int64  `json:"first_seen_ns"`
	LastSeenNs   int64  `json:"last_seen_ns"`
	Count        int64  `json:"count"`
	LastRelease  string `json:"last_release,omitempty"`
	LastDeployID string `json:"last_deploy_id,omitempty"`
	Status       string `json:"status"`
	Regressed    bool   `json:"regressed"`
}

ErrorIssue is one group of events as the list shows it.

type ErrorSample 

type ErrorSample struct {
	EventID     string `json:"event_id"`
	TimestampNs int64  `json:"timestamp_ns"`
	Release     string `json:"release,omitempty"`
	Environment string `json:"environment,omitempty"`
	DeployID    string `json:"deploy_id,omitempty"`
	Raw         string `json:"-"`
}

ErrorSample is one stored event of an issue.

type ErrorSpike 

type ErrorSpike struct {
	Start        time.Time  `json:"start"`
	End          time.Time  `json:"end"`
	Requests     float64    `json:"requests"`
	Errors5xx    float64    `json:"errors_5xx"`
	PeakRate5xx  float64    `json:"peak_error_rate_5xx"`
	BaselineRate float64    `json:"baseline_error_rate_5xx"`
	DeployID     string     `json:"deploy_id,omitempty"`
	DeployAt     *time.Time `json:"deploy_at,omitempty"`
	DeployStatus string     `json:"deploy_status,omitempty"`
	DeployImage  string     `json:"deploy_image,omitempty"`
	DeployCommit string     `json:"deploy_commit,omitempty"`
}

ErrorSpike is a run of consecutive buckets whose 5xx share stood out.

func DetectErrorSpikes 

func DetectErrorSpikes(points []RequestPoint, step time.Duration, deploys []DeployMarker, cfg SpikeConfig) []ErrorSpike

DetectErrorSpikes groups buckets of points (each step wide) whose 5xx share is at least cfg.MinRate, cfg.Factor times the window median and backed by cfg.MinRequests, then ties each spike to the latest deploy that started within cfg.Attribution before it. Newest spikes come first, capped at cfg.Max.

type Federator 

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

Federator fans a query out to every configured source and merges results, per ADR 008: "the control plane fans queries out to agents and merges results at query time; query is pull, not push." Today there is exactly one source (this node's own local DB), so federation is a merge-of-one; the shape is what Phase 3 fills in with real per-node sources.

func NewFederator 

func NewFederator(metrics []MetricsSource, logs []LogsSource) *Federator

NewFederator builds a Federator over an explicit source list, for Phase 3 once real per-node sources exist alongside (or instead of) the local one.

func NewLocalFederator 

func NewLocalFederator(db *DB) *Federator

NewLocalFederator builds a Federator with exactly one source: this process's own local DB. The single-node shape every control plane runs today, until Phase 3 adds real remote agents.

func (*Federator) LatestByMetric added in v0.2.0 

func (f *Federator) LatestByMetric(ctx context.Context, metric string) ([]Sample, error)

LatestByMetric is LatestByMetric's federated equivalent: fans out to every source and keeps, per resource_id, whichever source reported the newest sample. A resource split across sources (not possible today, single-node only, but Phase 3's per-node agents could each report a same-named resource) should never happen in practice, so "newest wins" is a defensive tie-break, not a real merge strategy.

func (*Federator) QueryAudience 

func (f *Federator) QueryAudience(ctx context.Context, resourceID string, from, to time.Time) ([]AudienceRow, error)

QueryAudience merges audience totals from every supporting source, tolerating partial failure like QueryUpstreams.

func (*Federator) QueryBreakdown added in v0.2.0 

func (f *Federator) QueryBreakdown(ctx context.Context, resourceID string, from, to time.Time) (Breakdown, error)

QueryBreakdown merges route and status totals from every source that implements BreakdownSource, tolerating partial failure like QueryMetrics.

func (*Federator) QueryLogs 

func (f *Federator) QueryLogs(ctx context.Context, resourceID string, from, to time.Time, query string) ([]LogEntry, error)

QueryLogs is QueryMetrics' log-query equivalent, merged and sorted by timestamp ascending the same way.

func (*Federator) QueryLogsLimited added in v0.2.0 

func (f *Federator) QueryLogsLimited(ctx context.Context, resourceID string, from, to time.Time, query string, limit int) ([]LogEntry, bool, error)

QueryLogsLimited is the federated bounded read: each source that supports it is capped at limit, the rest fall back to QueryLogs and are trimmed.

func (*Federator) QueryMetrics 

func (f *Federator) QueryMetrics(ctx context.Context, resourceID, metric string, from, to time.Time) ([]Sample, error)

QueryMetrics fans out to every metrics source and merges results, sorted by timestamp ascending. A source erroring doesn't fail the whole query: ADR 008's Consequences require the query layer to "handle partial results gracefully... deciding what the dashboard shows when [an agent is unreachable]," so this returns whatever the healthy sources answered, plus the errors as a joined error the caller can log or surface as a partial-result warning, not use to discard everything.

func (*Federator) QueryMetricsMany added in v0.2.0 

func (f *Federator) QueryMetricsMany(ctx context.Context, resourceIDs, metrics []string, from, to time.Time) ([]Sample, error)

QueryMetricsMany fans a batch read out to every source. A source without the batch method is read series by series. Errors from one source do not drop the others' samples, matching QueryMetrics.

func (*Federator) QueryRequestSamples 

func (f *Federator) QueryRequestSamples(ctx context.Context, resourceID string, from, to time.Time, limit int) ([]RequestSample, error)

QueryRequestSamples merges samples from every supporting source, newest first, trimmed to limit.

func (*Federator) QueryUpstreams 

func (f *Federator) QueryUpstreams(ctx context.Context, resourceID string, from, to time.Time) ([]UpstreamRow, error)

QueryUpstreams merges per-upstream totals from every source that supports them, tolerating partial failure like QueryBreakdown.

type FieldFilter added in v0.2.0 

type FieldFilter struct {
	Path  string
	Op    string
	Value string
}

FieldFilter matches one key of a structured (JSON) log line. Path may be dotted to reach nested objects.

func ParseFieldFilter added in v0.2.0 

func ParseFieldFilter(s string) (FieldFilter, error)

ParseFieldFilter parses "key=value", "key!=value", "key~text", "key>5".

func ParseFieldFilters added in v0.2.0 

func ParseFieldFilters(raw []string) ([]FieldFilter, error)

ParseFieldFilters parses several filters, rejecting more than the cap.

type HTTPSink 

type HTTPSink struct {
	URL    string
	Client *http.Client
}

HTTPSink POSTs a JSON array of httpSinkLine to URL for every Send call, the generic-webhook log drain shape (Coolify-parity item, see this package's drain.go doc comment).

func NewHTTPSink 

func NewHTTPSink(url string, client *http.Client) *HTTPSink

NewHTTPSink builds an HTTPSink. client defaults to a fresh *http.Client with httpSinkTimeout if nil.

func (*HTTPSink) Send 

func (s *HTTPSink) Send(ctx context.Context, entries []LogEntry) error

Send implements DrainSink.

type HostDiskCollector 

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

HostDiskCollector polls one filesystem path's capacity on an interval and writes disk_total_bytes/disk_used_bytes samples under one resource ID, the host-level counterpart to Collector's per-container polling above: a real reading of the volume that actually matters operationally (where images, volumes, and the SQLite stores live), not a per-container Docker stat.

func NewHostDiskCollector 

func NewHostDiskCollector(path, resourceID string, store *DB, interval time.Duration, logger *slog.Logger) *HostDiskCollector

NewHostDiskCollector builds a HostDiskCollector. logger defaults to slog.Default() if nil.

func (*HostDiskCollector) CollectOnce 

func (c *HostDiskCollector) CollectOnce(ctx context.Context) error

CollectOnce reads path's current capacity once and writes it.

func (*HostDiskCollector) Run 

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

Run calls CollectOnce every interval until ctx is done, the same "log and keep going, one bad tick must not stop the collector" shape Collector.Run above already establishes.

type HostMemoryCollector added in v0.2.0 

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

HostMemoryCollector polls this host's real memory capacity on an interval and writes memory_total_bytes/memory_available_bytes samples under one resource ID, HostDiskCollector's own memory counterpart. Like HostDiskCollector, it reads the machine the control plane process itself runs on, not every node in a multi-node fleet.

func NewHostMemoryCollector added in v0.2.0 

func NewHostMemoryCollector(resourceID string, store *DB, interval time.Duration, logger *slog.Logger) *HostMemoryCollector

NewHostMemoryCollector builds a HostMemoryCollector reading the platform's memory source (see HostMemoryBytes). logger defaults to slog.Default() if nil.

func (*HostMemoryCollector) CollectOnce added in v0.2.0 

func (c *HostMemoryCollector) CollectOnce(ctx context.Context) error

CollectOnce reads the host's current memory capacity once and writes it.

func (*HostMemoryCollector) Run added in v0.2.0 

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

Run calls CollectOnce every interval until ctx is done, the same "log and keep going, one bad tick must not stop the collector" shape HostDiskCollector.Run already establishes.

type HostPatchChecker 

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

HostPatchChecker looks up how many OS package updates a host has available, via whichever supported package manager exists on PATH.

func NewHostPatchChecker 

func NewHostPatchChecker() *HostPatchChecker

NewHostPatchChecker builds a HostPatchChecker that shells out to the real package manager binaries on PATH.

func (*HostPatchChecker) Check 

func (c *HostPatchChecker) Check(ctx context.Context) (PatchCounts, error)

Check runs the first supported package manager it finds (apt, then dnf, then yum) and returns its upgrade counts, or ErrNoSupportedPackageManager if none exist. A found manager whose command itself fails returns that error wrapped, never a false zero.

type HostPatchCollector 

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

HostPatchCollector polls one host's available OS package updates on an interval and writes MetricOSPatchesAvailable/ MetricOSSecurityPatchesAvailable samples under one resource ID, the same shape HostDiskCollector already establishes for host disk space.

func NewHostPatchCollector 

func NewHostPatchCollector(checker *HostPatchChecker, resourceID string, store *DB, interval time.Duration, logger *slog.Logger) *HostPatchCollector

NewHostPatchCollector builds a HostPatchCollector. checker defaults to NewHostPatchChecker() if nil, logger to slog.Default() if nil.

func (*HostPatchCollector) CollectOnce 

func (c *HostPatchCollector) CollectOnce(ctx context.Context) error

CollectOnce checks this host's available updates once and writes them. ErrNoSupportedPackageManager is not an error from this method's own perspective: it logs at Debug and returns nil without writing a sample, so a host with no supported package manager reports "unknown, not checked" (no recent sample) rather than a crashed collector or a false zero.

func (*HostPatchCollector) Run 

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

Run calls CollectOnce every interval until ctx is done, the same "log and keep going, one bad tick must not stop the collector" shape HostDiskCollector.Run already establishes.

type LimitedLogsSource added in v0.2.0 

type LimitedLogsSource interface {
	QueryLogsLimited(ctx context.Context, resourceID string, from, to time.Time, query string, limit int) (entries []LogEntry, truncated bool, err error)
}

LimitedLogsSource reads at most limit of the newest lines in a window.

type LoadBalancerSample added in v0.2.0 

type LoadBalancerSample struct {
	Service        string
	Total          int
	Healthy        int
	ActiveRequests int
}

LoadBalancerSample is one service's balancer snapshot.

type LocalStatsSource added in v0.2.0 

type LocalStatsSource struct {
	docker.StatsInspector
}

LocalStatsSource adapts a docker.StatsInspector (this process's own Docker client) into a StatsSource that ignores nodeID, for a single-node deployment with no remote agents to route to.

func (LocalStatsSource) Stats added in v0.2.0 

func (s LocalStatsSource) Stats(ctx context.Context, _, containerID string) (docker.ContainerStats, error)

Stats implements StatsSource by ignoring nodeID and calling the wrapped local inspector directly.

type LogBroadcaster 

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

LogBroadcaster fans out every log line LogCollector.StreamOne receives to whichever SSE viewers are currently watching that line's ResourceID, live, the moment it arrives. It exists for the same reason internal/deploylog.Recorder exists for build/deploy output: persistence alone (WriteLogBatch, batched up to logBatchMaxWait behind real time) is fine for a historical search but not for a view labeled "live", which needs to feel instant.

Deliberately simpler than deploylog.Recorder, in one specific way: there is no Start/Finish lifecycle here. A deploy attempt has a real bounded shape (started, eventually finishes, gets evicted), a container's ResourceID does not: a service can be redeployed, restarted, or have zero subscribers for most of its life, with no single moment that means "this resource is done, stop remembering it." So LogBroadcaster never eagerly allocates state for a resource, only when something actually calls Subscribe, and Unsubscribe cleans up empty entries rather than relying on a Finish call that has no natural trigger here. There is also no "lines so far" in-memory replay (contrast Recorder.Snapshot): that job belongs to the persisted store (QueryLogs) via internal/api's live-log handler, which backfills recent context from there before calling Subscribe, since only the store, not this in-memory broadcaster, survives a control plane restart.

One instance is shared across every LogCollector.StreamOne goroutine (the publishing side, one per currently-running container) and every live-log SSE handler request (the subscribing side), the same "exactly one shared instance, wired once at startup" shape deploylog.Recorder's own package doc comment establishes for build logs: a viewer connected through the HTTP API must see the same lines the collector is receiving directly from Docker, which only works if both sides publish to and read from the same subscriber set.

func NewLogBroadcaster 

func NewLogBroadcaster() *LogBroadcaster

NewLogBroadcaster builds an empty LogBroadcaster.

func (*LogBroadcaster) Publish 

func (b *LogBroadcaster) Publish(entry LogEntry)

Publish delivers entry to every subscriber currently watching entry.ResourceID. Non-blocking by design: a subscriber whose channel is full (see broadcastBufferSize) is skipped for this entry rather than stalling the call, since Publish's real caller (StreamOne) is on the same goroutine that is also consuming Docker's own log stream and batching writes to the store, neither of which may block on an HTTP response writer somewhere downstream that a slow or stuck client left half-read.

func (*LogBroadcaster) Subscribe 

func (b *LogBroadcaster) Subscribe(resourceID string) (ch <-chan LogEntry, unsubscribe func())

Subscribe registers a new live listener for resourceID, returning a channel that receives every entry Published for that resourceID from this point forward, and an unsubscribe func the caller must call exactly once, typically deferred, once it stops reading (e.g. the SSE client disconnected). Safe to call for a resourceID with no active producer (the container hasn't started yet, or already stopped, or this control plane never ran a collector for it): Subscribe never fails, it may simply never receive anything, which is the correct behavior for a live tail opened against a currently quiet or not-yet-running app rather than an error condition.

func (*LogBroadcaster) SubscribeAll added in v0.2.0 

func (b *LogBroadcaster) SubscribeAll() (ch <-chan LogEntry, unsubscribe func())

SubscribeAll returns a channel receiving every resource's lines, with the same non-blocking drop-on-slow-subscriber behavior as Subscribe.

func (*LogBroadcaster) SubscribeCounted 

func (b *LogBroadcaster) SubscribeCounted(resourceID string) (ch <-chan LogEntry, takeDropped func() int64, unsubscribe func())

SubscribeCounted is Subscribe plus takeDropped, which returns how many lines were skipped for this subscriber since the last call and resets it.

type LogBudgetTrim 

type LogBudgetTrim struct {
	ResourceID string
	Lines      int64
	Bytes      int64
}

LogBudgetTrim reports what one enforcement pass removed for one resource.

type LogCollector 

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

LogCollector streams logs from a caller-supplied set of targets and writes what it receives to a Store, batching writes per logBatchMaxLines/logBatchMaxWait, and, when a broadcaster is configured, fans every line out live at the same time (see StreamOne).

func NewLogCollector 

func NewLogCollector(source LogSource, store *DB, broadcaster *LogBroadcaster, logger *slog.Logger) *LogCollector

NewLogCollector builds a LogCollector. broadcaster may be nil: without one, StreamOne still batches and persists every line exactly as before, it just has nowhere to publish a live copy, so a control plane started without one simply has no live log tail (internal/api's live log route returns 501 in that case, the same "not configured" shape this package's own optional dependencies already use elsewhere). logger defaults to slog.Default() if nil.

func (*LogCollector) Run 

func (lc *LogCollector) Run(ctx context.Context, resync time.Duration, targetsFunc func(context.Context) ([]LogTarget, error)) error

Run derives targets via targetsFunc on a resync interval and keeps exactly one StreamOne goroutine alive per currently-desired container ID, starting new ones as containers appear and cancelling ones for containers no longer desired, until ctx is done. This is the streaming-collector equivalent of Collector.Run's "re-derive targets every tick" principle: the set of things being collected stays current as services are created, redeployed, or removed, without needing a separate mechanism to notice that a target disappeared mid-stream versus never existed.

func (*LogCollector) StreamOne 

func (lc *LogCollector) StreamOne(ctx context.Context, target LogTarget) error

StreamOne follows target's log stream (Follow: true, from the current point forward) until ctx is cancelled or the stream ends on its own (the container stopped), batching lines per logBatchMaxLines/ logBatchMaxWait and writing each batch to the store. Blocking: callers run it in its own goroutine per target, one long-lived subscription per running container, not a polling tick like Collector.CollectOnce: logs are a push stream from Docker, not a point-in-time snapshot, so this package's two collectors have genuinely different shapes even though they share a store and a retention pattern.

func (*LogCollector) Wake 

func (lc *LogCollector) Wake()

Wake asks Run to re-derive targets now, e.g. on a container start or die event, so a container shorter-lived than the resync tick is seen.

type LogEntry 

type LogEntry struct {
	ResourceID string
	// ContainerID is the container this line actually came from. Empty
	// for rows written before migrations/0005_log_entries_container_id.sql
	// existed; never empty for a line written by this version onward (see
	// toLogEntry). This is what lets a live view tell a dying old
	// container's lines apart from the new, current one's during a
	// blue-green overlap, see LogCollector.Run's own doc comment.
	ContainerID string
	// Stream is "stdout" or "stderr".
	Stream    string
	Timestamp time.Time
	Message   string
	// Structured is true when Message's full text parsed as a JSON
	// object; see classifyLine.
	Structured bool
	// FieldsJSON holds Message's parsed JSON when Structured is true,
	// stored in its own column, separately from the raw line, so a future
	// log viewer doesn't need
	// to re-parse Message to tell structured and plain lines apart or to
	// render one differently from the other. Empty when Structured is
	// false.
	FieldsJSON string
}

LogEntry is one line of container log output, landed in or read back from the store.

func QueryLogsBounded added in v0.2.0 

func QueryLogsBounded(ctx context.Context, src BoundedLogQuerier, resourceID string, from, to time.Time, query string, limit int) ([]LogEntry, bool, error)

QueryLogsBounded reads at most limit of the newest lines from src, using the source's bounded read when it has one and trimming otherwise.

func RankLogsByRelevance 

func RankLogsByRelevance(entries []LogEntry, query string) []LogEntry

RankLogsByRelevance orders entries, which already matched query, by BM25 score (best first), newest first among equal scores. Terms under NOT or -word never score. The corpus is the matched set, so scores compare lines within one result, not across searches.

type LogQLLineFilter 

type LogQLLineFilter struct {
	Op    string
	Value string
	// contains filtered or unexported fields
}

LogQLLineFilter is one |= != |~ !~ line filter.

func (LogQLLineFilter) Matches 

func (f LogQLLineFilter) Matches(line string) bool

Matches reports whether the line passes the filter.

type LogQLMatcher 

type LogQLMatcher struct {
	Name  string
	Op    string
	Value string
	// contains filtered or unexported fields
}

LogQLMatcher is one {label op "value"} selector term.

func (LogQLMatcher) Matches 

func (m LogQLMatcher) Matches(value string) bool

Matches reports whether the label value satisfies the matcher.

type LogQLQuery 

type LogQLQuery struct {
	Matchers []LogQLMatcher
	Filters  []LogQLLineFilter
}

LogQLQuery is a parsed stream selector plus line filters.

func ParseLogQL 

func ParseLogQL(src string) (LogQLQuery, error)

ParseLogQL parses the supported LogQL subset: a stream selector on app, stream and level followed by line filters. Everything else is an error rather than a quietly wrong answer.

func (LogQLQuery) MatchApp 

func (q LogQLQuery) MatchApp(app string) bool

MatchApp applies only the app terms, used to pick which apps to read.

func (LogQLQuery) MatchLabels 

func (q LogQLQuery) MatchLabels(labels map[string]string) bool

MatchLabels reports whether every selector term holds for the label set.

func (LogQLQuery) MatchLine 

func (q LogQLQuery) MatchLine(line string) bool

MatchLine applies every line filter in order.

type LogSource 

type LogSource interface {
	Logs(ctx context.Context, containerID string, follow bool, since time.Time) (<-chan docker.LogLine, <-chan error)
}

LogSource is the narrow Docker surface a LogCollector needs: nothing about container discovery, only "give me a demultiplexed log stream for this ID." *docker.Client satisfies this structurally. Deliberately not part of docker.Runtime, the same reasoning StatsSource's doc comment already gives for the same choice: adding this to Runtime would mean updating every existing fake Runtime implementation across the codebase for a capability only this package needs.

type LogStreamResult added in v0.2.0 

type LogStreamResult struct {
	Lines int64
	// EndNs is the exclusive upper bound actually covered; it is below the
	// requested bound only when the line cap truncated the range.
	EndNs int64
}

LogStreamResult describes how far StreamLogs got.

type LogTarget 

type LogTarget struct {
	ResourceID  string
	ContainerID string
	// Stopped marks an exited container: its output is fetched once
	// (non-follow) so a short-lived or crashed container is not lost.
	Stopped bool
}

LogTarget is one container a LogCollector should stream logs from.

type LogUsage 

type LogUsage struct {
	ResourceID   string
	Bytes        int64
	Lines        int64
	RatePerMin   float64 // bytes per minute averaged over the last 24h
	BudgetBytes  int64   // 0 means unlimited
	BudgetCustom bool    // true when a per-app override is stored
	TrimmedLines int64
	TrimmedBytes int64
}

LogUsage is one resource's stored log volume against its budget.

type LogsSource 

type LogsSource interface {
	QueryLogs(ctx context.Context, resourceID string, from, to time.Time, query string) ([]LogEntry, error)
}

LogsSource is the log-query equivalent of MetricsSource.

type Maintainer added in v0.2.0 

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

Maintainer runs rollups and pruning on timers.

func NewMaintainer added in v0.2.0 

func NewMaintainer(db *DB, rollupEvery, pruneEvery time.Duration, logger *slog.Logger) *Maintainer

NewMaintainer builds a Maintainer; zero intervals default to 1 minute for rollups and 1 hour for pruning.

func (*Maintainer) PruneOnce added in v0.2.0 

func (m *Maintainer) PruneOnce(ctx context.Context) error

PruneOnce runs Prune and logs what it removed.

func (*Maintainer) RollupOnce added in v0.2.0 

func (m *Maintainer) RollupOnce(ctx context.Context) error

RollupOnce runs Rollup, repeating (bounded) while a tier has backlog.

func (*Maintainer) Run added in v0.2.0 

func (m *Maintainer) Run(ctx context.Context)

Run blocks until ctx is cancelled, rolling up and pruning on their timers.

type ManyMetricsQuerier added in v0.2.0 

type ManyMetricsQuerier interface {
	QueryMetricsMany(ctx context.Context, resourceIDs, metrics []string, from, to time.Time) ([]Sample, error)
}

ManyMetricsQuerier is the federated counterpart of ManyMetricsSource.

type ManyMetricsSource added in v0.2.0 

type ManyMetricsSource interface {
	QueryMany(ctx context.Context, resourceIDs, metrics []string, from, to time.Time) ([]Sample, error)
}

ManyMetricsSource is the optional batch read a MetricsSource can add: every (resource, metric) series in one round trip instead of one query per series.

type MetricsQuerier added in v0.2.0 

type MetricsQuerier interface {
	QueryMetrics(ctx context.Context, resourceID, metric string, from, to time.Time) ([]Sample, error)
}

MetricsQuerier is the read surface the request queries need; *Federator satisfies it, so request metrics federate exactly like container metrics.

type MetricsSource 

type MetricsSource interface {
	Query(ctx context.Context, resourceID, metric string, from, to time.Time) ([]Sample, error)
	LatestByMetric(ctx context.Context, metric string) ([]Sample, error)
}

MetricsSource is what the query layer needs from one agent, local or remote, to answer a metrics query. *DB satisfies this today (the single-node, in-process "agent"); a Phase 3 gRPC client would satisfy it too without Federator's callers (internal/api) ever changing. This mirrors the same precedent already set for the reconcile agent transport: it must also work in single-node mode, communicating over an in-memory transport that implements the same interface.

type NewSignatureDetector added in v0.2.0 

type NewSignatureDetector struct {
	Src BoundedLogQuerier
	// MaxRows bounds each of the two reads.
	MaxRows int
}

NewSignatureDetector counts error signatures that showed up after a deploy and did not exist in the equally long window before it.

func (NewSignatureDetector) NewErrorTypes added in v0.2.0 

func (d NewSignatureDetector) NewErrorTypes(ctx context.Context, app string, deployedAt, until time.Time) (int, error)

NewErrorTypes compares [deployedAt, until] (clamped to between ten minutes and a day) with the same span before deployedAt.

type PatchCounts 

type PatchCounts struct {
	Manager  string
	Total    int
	Security int
}

PatchCounts is one point-in-time reading of a host's available OS package updates.

type PipelineRule 

type PipelineRule struct {
	Kind    string `json:"kind"`
	Pattern string `json:"pattern"`
	Replace string `json:"replace,omitempty"`
}

PipelineRule is one step. multiline: Pattern matches the first line of a record, other lines are joined to it. redact: matches are replaced. parse: named groups become structured fields. drop: matching lines go.

type PreviewLine 

type PreviewLine struct {
	Message string            `json:"message"`
	Dropped bool              `json:"dropped"`
	Fields  map[string]string `json:"fields,omitempty"`
}

PreviewLine is the outcome of one sample line.

func PreviewPipeline 

func PreviewPipeline(rules []PipelineRule, lines []string) ([]PreviewLine, error)

PreviewPipeline runs rules over sample lines the way the collector does.

type PruneResult added in v0.2.0 

type PruneResult struct {
	Raw, Minute, Hour int64
	// OverCap is true when the size cap is still exceeded after pruning.
	OverCap bool
}

PruneResult reports what one Prune run removed.

type RequestDrainer added in v0.2.0 

type RequestDrainer interface {
	DrainByApp() map[string]RequestWindow
}

RequestDrainer hands over and resets request activity accumulated since the last call, keyed by app name. *ingress.RequestStats satisfies it.

type RequestOverview added in v0.2.0 

type RequestOverview struct {
	Summary RequestSummary
	Points  []RequestPoint
}

RequestOverview is one app's request summary over a window plus its per step rate series, both derived from one shared sample read.

type RequestPoint added in v0.2.0 

type RequestPoint struct {
	Timestamp      time.Time `json:"timestamp"`
	Requests       float64   `json:"requests"`
	RatePerSec     float64   `json:"rate_per_sec"`
	ErrorRate4xx   float64   `json:"error_rate_4xx"`
	ErrorRate5xx   float64   `json:"error_rate_5xx"`
	UpstreamErrors float64   `json:"upstream_errors"`
	P50Ms          float64   `json:"p50_ms"`
	P95Ms          float64   `json:"p95_ms"`
	P99Ms          float64   `json:"p99_ms"`
	BytesInPerSec  float64   `json:"bytes_in_per_sec"`
	BytesOutPerSec float64   `json:"bytes_out_per_sec"`
}

RequestPoint is one step of an app's request series. Rates are per second over the step; error rates are fractions of that step's requests.

func QueryRequests added in v0.2.0 

func QueryRequests(ctx context.Context, q MetricsQuerier, app string, from, to time.Time, step time.Duration) ([]RequestPoint, error)

QueryRequests returns the app's request series over [from, to] bucketed by step (at least MinRequestStep). Buckets with no traffic are omitted. Counters are summed across sources, so a multi-node app merges correctly.

type RequestSample 

type RequestSample struct {
	Time       time.Time `json:"time"`
	Route      string    `json:"route"`
	Method     string    `json:"method"`
	Status     int       `json:"status"`
	DurationMS float64   `json:"duration_ms"`
	Upstream   string    `json:"upstream,omitempty"`
	RequestID  string    `json:"request_id,omitempty"`
}

RequestSample is one slow or failed request. Route is a normalized template, never a raw path.

type RequestSampler added in v0.2.0 

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

RequestSampler writes drained request activity as samples on a fixed tick.

func NewRequestSampler added in v0.2.0 

func NewRequestSampler(src RequestDrainer, db *DB, interval time.Duration, logger *slog.Logger) *RequestSampler

NewRequestSampler builds a sampler; a nil logger falls back to slog.Default.

func (*RequestSampler) Run added in v0.2.0 

func (s *RequestSampler) Run(ctx context.Context)

Run samples every interval until ctx is cancelled.

func (*RequestSampler) SampleOnce added in v0.2.0 

func (s *RequestSampler) SampleOnce(ctx context.Context, at time.Time) error

SampleOnce drains the source and stores the result at time at.

type RequestSummary added in v0.2.0 

type RequestSummary struct {
	WindowSeconds  float64 `json:"window_seconds"`
	HasTraffic     bool    `json:"has_traffic"`
	Requests       float64 `json:"requests"`
	RatePerSec     float64 `json:"rate_per_sec"`
	ErrorRate4xx   float64 `json:"error_rate_4xx"`
	ErrorRate5xx   float64 `json:"error_rate_5xx"`
	P95Ms          float64 `json:"p95_ms"`
	UpstreamErrors float64 `json:"upstream_errors"`
}

RequestSummary is an app's request health over one window.

func SummarizeRequests added in v0.2.0 

func SummarizeRequests(ctx context.Context, q MetricsQuerier, app string, window time.Duration, now time.Time) (RequestSummary, error)

SummarizeRequests reduces the app's last window of traffic to one summary, the entry point auto-rollback and SLO features build on. A window with no requests yields HasTraffic false and zero values.

type RequestWindow added in v0.2.0 

type RequestWindow struct {
	Requests       uint64
	Status2xx      uint64
	Status3xx      uint64
	Status4xx      uint64
	Status5xx      uint64
	UpstreamErrors uint64
	BytesIn        uint64
	BytesOut       uint64
	Latency        [LatencyBucketCount]uint64
	// Shield counts requests the abuse shield turned away, by reason.
	Shield [ShieldReasonCount]uint64
	// Abuse counts the per-app ingress abuse defense decisions.
	Abuse AbuseWindow
	// Routes and Statuses are the bounded per-route and exact-status
	// breakdown; they are stored in request_breakdown, not as samples.
	Routes   map[string]RouteStat
	Statuses map[int]uint64
	// Upstreams and Sampled are the bounded per-replica histograms and the
	// slow or failed request sample, stored in their own tables.
	Upstreams map[string]UpstreamStat
	Sampled   []RequestSample
	// Referrers and Countries are bounded label counts stored in
	// request_audience.
	Referrers map[string]uint64
	Countries map[string]uint64
}

RequestWindow is one host key's request activity over one sampling tick.

func (*RequestWindow) Add added in v0.2.0 

func (w *RequestWindow) Add(o RequestWindow)

Add merges o into w.

func (*RequestWindow) AddSample 

func (w *RequestWindow) AddSample(s RequestSample, tickMax int)

AddSample keeps s unless the window already holds tickMax samples.

func (*RequestWindow) ObserveAudience 

func (w *RequestWindow) ObserveAudience(referrerHost, country string, limit int)

ObserveAudience counts one request under its referrer host (ReferrerDirect when empty) and, when known, its ISO country code.

func (*RequestWindow) ObserveRoute added in v0.2.0 

func (w *RequestWindow) ObserveRoute(route string, status int, latencyMS float64, routeMax int)

ObserveRoute records one request into the window's route and status maps.

func (*RequestWindow) ObserveUpstream 

func (w *RequestWindow) ObserveUpstream(upstream string, status int, upstreamErr bool, latencyMS float64, limit int)

ObserveUpstream records one request against its upstream label; labels beyond max distinct values per window collapse into OverflowUpstream.

func (RequestWindow) Samples added in v0.2.0 

func (w RequestWindow) Samples(resourceID string, at time.Time) []Sample

Samples renders the window as non-zero metric samples for resourceID.

type RollupResult added in v0.2.0 

type RollupResult struct {
	Buckets int
	// More is true when a tier stopped at MaxBucketsPerRun with work left.
	More bool
}

RollupResult reports one Rollup run.

type RouteRow added in v0.2.0 

type RouteRow struct {
	Route        string
	Requests     float64
	Errors4xx    float64
	Errors5xx    float64
	LatencyMSSum float64
}

RouteRow is one route's totals over a query window.

type RouteStat added in v0.2.0 

type RouteStat struct {
	Requests     uint64
	Errors4xx    uint64
	Errors5xx    uint64
	LatencyMSSum float64
}

RouteStat is one route's counters over a sampling tick.

type Sample 

type Sample struct {
	ResourceID string
	Metric     string
	Timestamp  time.Time
	Value      float64
}

Sample is one metric reading at one point in time for one resource (e.g. resource_id "service:web", metric "cpu_percent").

func SumAcrossResources 

func SumAcrossResources(nodeResourceID, metric string, samples []Sample) []Sample

SumAcrossResources collapses samples from multiple resources (e.g. every container currently placed on one node) for a single metric into one per-timestamp total: "how much of this metric was in use, summed across everything running here, at each collection tick."

Samples are grouped by their exact Timestamp, not a time bucket: Collector.CollectOnce (collector.go) captures one `now` per tick and stamps every target polled in that tick with it, and WriteSamples/Query round-trip that value at one-second precision (store.go), so two containers collected in the same tick share the exact same Timestamp here, not merely an approximately-close one. A caller that wants coarser buckets afterward (a chart's `step` query param) runs this function's output through Aggregate, the same as a single resource's own series already does (internal/api/metrics.go's handleQueryMetrics).

Mixing metrics in one call would produce a meaningless sum (adding a percent to a byte count), so this assumes every sample passed in already shares one metric; callers query one metric per resource and concatenate before calling this, matching Federator.QueryMetrics' own "one metric per call" contract.

The returned samples' ResourceID and Metric are the caller-supplied nodeResourceID/metric, not copied from any individual input sample: the result represents a virtual, computed-on-the-fly aggregate resource, not any single container, and is never itself written back to the store.

Two known, deliberately accepted limitations, not bugs:

  • The exact-Timestamp grouping this function relies on holds only because everything collected today runs through one process's own Collector.CollectOnce, sharing one clock. The Phase 3 plan (per-node agents, each with an independent local collector) breaks this assumption: two agents' clocks are never perfectly synchronized, so their samples for "the same tick" won't share an identical Unix second. Revisit this grouping (a tolerance window, not exact equality) before this function is ever fed cross-agent samples, not after.
  • A bucket is never flagged as partially-covered: if one resource has a sample at a given timestamp and another (also placed on this node for that whole range) doesn't, the sum silently reflects only the resource that reported, indistinguishable here from "every resource genuinely totaled this." Neither zero-filled nor crashed, both worse options, but a caller cannot currently tell "real total" from "partial total" from this function's output alone.

type SampleConfig 

type SampleConfig struct {
	Rate    float64
	SlowMS  float64
	TickMax int
}

SampleConfig decides which requests enter the sample log.

func SampleConfigFromEnv 

func SampleConfigFromEnv() SampleConfig

SampleConfigFromEnv reads the sampling knobs, falling back to defaults on unset or invalid values.

func (SampleConfig) ShouldSample 

func (c SampleConfig) ShouldSample(status int, durationMS float64, upstreamErr bool, requestID string) bool

ShouldSample reports whether a finished request belongs in the sample log: failures always, slow requests at Rate. The decision hashes the request id so it is stable and needs no random source.

type Signature added in v0.2.0 

type Signature struct {
	ID        string    `json:"id"`
	Template  string    `json:"template"`
	Level     string    `json:"level"`
	Count     int       `json:"count"`
	FirstSeen time.Time `json:"first_seen"`
	LastSeen  time.Time `json:"last_seen"`
	// Buckets is the per-interval count across the query window, for a sparkline.
	Buckets []int `json:"buckets"`
	// Sample is the newest line of the cluster, redacted and truncated.
	Sample string `json:"sample"`
	// New is set when the signature has no occurrence before the baseline cut.
	New bool `json:"new,omitempty"`
}

Signature is one cluster of similar log lines.

func AggregateSignatures added in v0.2.0 

func AggregateSignatures(entries []LogEntry, from, to time.Time, opts SignatureOptions) []Signature

AggregateSignatures clusters entries inside [from, to] by signature, most frequent first (ties by most recent).

type SignatureOptions added in v0.2.0 

type SignatureOptions struct {
	// MinLevel keeps lines at or above this level; "" keeps every line.
	MinLevel string
	// Buckets is the sparkline resolution; zero means 24.
	Buckets int
	// Top caps the result; zero means no cap.
	Top int
}

SignatureOptions tunes AggregateSignatures.

type Span 

type Span struct {
	TraceID       string
	SpanID        string
	ParentID      string
	ResourceID    string
	Name          string
	Kind          int
	StartNs       int64
	EndNs         int64
	StatusCode    int // 0 unset, 1 ok, 2 error
	StatusMessage string
	Attrs         string
}

Span is one trace span as stored: identifiers are lowercase hex, times are unix nanoseconds, and attrs is the span's attributes as a JSON object.

type SpikeConfig 

type SpikeConfig struct {
	MinRate     float64
	Factor      float64
	MinRequests float64
	Attribution time.Duration
	Max         int
	Buckets     int
}

SpikeConfig tunes error spike detection. MinRate is the 5xx share a bucket must reach, Factor how many times the window median it must exceed, and MinRequests the traffic a bucket needs so a lone failure is not a spike.

func SpikeConfigFromEnv 

func SpikeConfigFromEnv() SpikeConfig

SpikeConfigFromEnv reads the detection thresholds from the environment.

type StatsSource 

type StatsSource interface {
	Stats(ctx context.Context, nodeID, containerID string) (docker.ContainerStats, error)
}

StatsSource is the narrow Docker surface a Collector needs: a snapshot for one container on one node. Deliberately not part of docker.Runtime (would ripple into every existing fake, the same reasoning internal/reconcile/application's ServiceStore already gives); cmd/levelrail/main.go's multiNodeStatsSource routes local vs. remote nodes behind this one interface.

type StatusRow added in v0.2.0 

type StatusRow struct {
	Status int
	Count  float64
}

StatusRow is one exact status code's total over a query window.

type SyslogSink 

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

SyslogSink writes each entry to a syslog daemon via the standard library's log/syslog package, which this build tag mirrors exactly: that package itself only builds on the platforms listed above. This platform targets Linux-only nodes in production, so the unsupported set (Windows, Plan 9, WASM) is not a real gap for a managed node, only for compiling this package on an unsupported dev machine, which the sibling stub file (drain_syslog_stub.go) covers.

func (*SyslogSink) Send 

func (s *SyslogSink) Send(ctx context.Context, entries []LogEntry) error

Send implements DrainSink. log/syslog's Writer takes no context, so this only checks ctx before each line rather than mid-write, enough to stop promptly on shutdown without needing a custom transport.

type Target 

type Target struct {
	ResourceID  string
	ContainerID string
	// NodeID is resolveNodeTransport's own "" means local convention
	// (cmd/levelrail/main.go): which node ContainerID actually runs on,
	// so a StatsSource covering more than one node can route each poll
	// correctly without Target carrying a whole Transport.
	NodeID string
}

Target is one container a Collector should poll on each tick. ResourceID is the metrics store's own identifier for whatever this container belongs to (e.g. "service:web"), deliberately not the raw container ID: a redeploy gives a service a new container ID (per internal/reconcile/application's deterministic-name-per-image design) but its metric history should stay queryable under one stable identifier across that change.

type TierConfig added in v0.2.0 

type TierConfig struct {
	RawRetention    time.Duration
	MinuteRetention time.Duration
	HourRetention   time.Duration
	// RawMaxRange and MinuteMaxRange are the widest query ranges served
	// from the raw and 1 minute tiers; wider ranges use the next tier.
	RawMaxRange    time.Duration
	MinuteMaxRange time.Duration
	// MaxBytes caps the live size of the database file; 0 disables it.
	MaxBytes int64
	// Lag delays closing a bucket so late samples still land in it.
	Lag time.Duration
	// MaxBucketsPerRun bounds the buckets one rollup run computes per tier.
	MaxBucketsPerRun int
}

TierConfig controls rollup tiers, retention and query tier selection.

func DefaultTierConfig added in v0.2.0 

func DefaultTierConfig() TierConfig

DefaultTierConfig returns the documented defaults.

func TierConfigFromEnv added in v0.2.0 

func TierConfigFromEnv(getenv func(string) string) TierConfig

TierConfigFromEnv reads the APP_METRICS_* overrides through getenv and falls back to DefaultTierConfig for unset or invalid values.

type TraceFilter 

type TraceFilter struct {
	ResourceID string
	From, To   time.Time
	MinMillis  int64
	ErrorsOnly bool
	Limit      int
}

TraceFilter narrows ListTraces.

type TraceSummary 

type TraceSummary struct {
	TraceID    string `json:"trace_id"`
	RootName   string `json:"root_name"`
	StartNs    int64  `json:"start_ns"`
	DurationNs int64  `json:"duration_ns"`
	Spans      int    `json:"spans"`
	Errors     int    `json:"errors"`
}

TraceSummary is one trace as the list shows it.

type TrafficDetailSource 

type TrafficDetailSource interface {
	QueryUpstreams(ctx context.Context, resourceID string, from, to time.Time) ([]UpstreamRow, error)
	QueryRequestSamples(ctx context.Context, resourceID string, from, to time.Time, limit int) ([]RequestSample, error)
}

TrafficDetailSource reads per-upstream totals and sampled requests; *DB satisfies it.

type UpstreamRow 

type UpstreamRow struct {
	Upstream       string
	Requests       float64
	Errors4xx      float64
	Errors5xx      float64
	UpstreamErrors float64
	Latency        []float64
}

UpstreamRow is one upstream's totals and latency histogram over a window.

type UpstreamStat 

type UpstreamStat struct {
	Requests       uint64
	Errors4xx      uint64
	Errors5xx      uint64
	UpstreamErrors uint64
	Latency        [LatencyBucketCount]uint64
}

UpstreamStat is one upstream's counters over a sampling tick.

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.